javaapache-kafkakafka-consumer-api

How to fetch messages which are uncommited in kafka

I have a function in java in which I am trying to fetch messages which are unread. For example, If I have messages with offSet 0,1,2 in broker which are already read by the consumer and If I switch off my consumer for an hour. And at that time I produce messages with offset 3,4,5. After that when my consumer is started it should read message from offset 3 not from 0. But, It either reads all the messages or read those messages which are produced after starting Kafka consumer. I want to read those messages which are unread or uncommited

I tried "auto.offset.reset"= "latest" and "earliest". as well as "enable.auto.commit" = "true" and "false". I also tried commitSync() and commitAsync() before calling close() method but no luck.

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 just want to fetch only unread messages in kafka Consumer. Please correct me if I am wrong somewhere. And Thanks in Advance

Solution

  • The main problem in your code is that you are not closing the consumer you used to poll messages; this is because each call to createConsumer() creates a new KafkaConsumer. And as you are not closing the consumer, and are calling poll() only once, you never commit the messages you have read.

    (with auto-commit, commit is called within poll() after auto-commit-interval, and within close())

    Once you will have corrected that it should work with following settings: