Как получить незафиксированные сообщения в Kafka ⇐ JAVA

[Anonymous]

У меня есть функция в Java, с помощью которой я пытаюсь получить непрочитанные сообщения. Например, если у меня есть сообщения с offSet 0,1,2 в брокере, которые уже прочитаны потребителем, и если я отключу своего потребителя на час. И в это время я создаю сообщения со смещением 3,4,5. После этого, когда мой потребитель запускается, он должен читать сообщение со смещением 3, а не со смещением 0. Но он либо читает все сообщения, либо читает те сообщения, которые создаются после запуска потребителя Kafka. Я хочу прочитать непрочитанные или незафиксированные сообщения

Я попробовал «auto.offset.reset» = «последние» и «самые ранние». а также "enable.auto.commit" = "true" и "false". Я также попробовал commitSync() и commitAsync() перед вызовом метода close(), но безуспешно.

Код: Выделить всё

public static KafkaConsumer createConsumer() {

Properties properties = new Properties();
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, Constants.KAFKA_BROKER);
properties.put(ConsumerConfig.GROUP_ID_CONFIG, "testGroup");
properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
properties.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "50");
properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1");
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

KafkaConsumer consumer = new KafkaConsumer(properties);

consumer.subscribe(Collections.singleton(Constants.TOPIC));
return consumer;

}

public static void main(String[] args) {

Код: Выделить всё

    System.out.println("");
System.out.println("----------------");
System.out.println("");
System.out.println("KAFKA CONSUMER EXAMPLE");
System.out.println("");
System.out.println("----------------");
System.out.println("");

OffsetAndMetadata offsetAndMetadataInitial = createConsumer().committed(new TopicPartition(Constants.TOPIC, 0));

System.out.println("");
System.out.println("Offset And MetaData Initial : ");
System.out.println(offsetAndMetadataInitial);
System.out.println("");

ConsumerRecords consumerRecords = createConsumer().poll(Duration.ofSeconds(2L));

System.out.println("");
System.out.println("Count Consumer Records : " + consumerRecords.count());
System.out.println("");

Iterator itr = consumerRecords.iterator();
Map partationOffsetMap = new HashMap(4);

while (itr.hasNext()) {

ConsumerRecord record = itr.next();

System.out.println("OffSet : " + record.offset());
System.out.println("");
System.out.println("Key : " + record.key());
System.out.println("Value : " + record.value());
System.out.println("Partition : " + record.partition());
System.out.println("--------------------------");
System.out.println("");

}

createConsumer().close();

}

Я просто хочу получить только непрочитанные сообщения в Kafka Consumer. Поправьте меня, пожалуйста, если я где-то ошибаюсь. И заранее спасибо

Подробнее здесь: https://stackoverflow.com/questions/566 ... d-in-kafka