Как получить сообщения, которые не передаются в кафку - PullRequest
Новая
       2

Как получить сообщения, которые не передаются в кафку

1 голос
/ 17 июня 2019

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

Я пробовал "auto.offset.reset"= "latest" и "earliest".а также "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<String, String> consumer = new KafkaConsumer<String, String>(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<String, String> consumerRecords = createConsumer().poll(Duration.ofSeconds(2L));

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

    Iterator<ConsumerRecord<String, String>> itr = consumerRecords.iterator();
    Map<TopicPartition, OffsetAndMetadata> 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();

}

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

1 Ответ

3 голосов
/ 17 июня 2019

Основная проблема в вашем коде заключается в том, что вы не закрываете потребителя, которого вы использовали для опроса сообщений; Это происходит потому, что каждый вызов createConsumer () создает новый KafkaConsumer. И поскольку вы не закрываете потребителя и вызываете poll () только один раз, вы никогда не фиксируете прочитанные сообщения.

(при автоматической фиксации коммит вызывается в poll () после интервала автоматической фиксации и в close ())

После того, как вы исправите, он должен работать со следующими настройками:

  • auto-commit = true (в противном случае вы можете также зафиксировать вручную, но автоматическое принятие проще).
  • смещение-сброс = самое раннее (это влияет только на первый раз, когда вы потребляете для данного идентификатора группы, чтобы сказать, если вы хотите использовать с начала темы или только сообщения, созданные после того, как вы начали потреблять. Как только вы начали использовать с заданным идентификатором группы, вы всегда будете продолжать использовать после последнего принятого вами смещения.)
  • идентификатор группы не должен изменяться между перезапусками, иначе вы начнете с начала или с конца снова, в зависимости от настроек сброса смещения.

Надеюсь, это поможет

...