title: How to fetch messages which are uncommited in kafka
body: <p>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</p>
  
  <p>I tried <code>"auto.offset.reset"= "latest"</code> and <code>"earliest"</code>. as well as <code>"enable.auto.commit" = "true"</code> and <code>"false"</code>. I also tried <code>commitSync()</code> and <code>commitAsync()</code> before calling <code>close()</code> method but no luck.</p>
  
  <pre><code>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&lt;String, String&gt; consumer = new KafkaConsumer&lt;String, String&gt;(properties);
  
      consumer.subscribe(Collections.singleton(Constants.TOPIC));
      return consumer;
  
  }
  </code></pre>
  
  <p>public static void main(String[] args) {</p>
  
  <pre><code>    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&lt;String, String&gt; consumerRecords = createConsumer().poll(Duration.ofSeconds(2L));
  
      System.out.println("");
      System.out.println("Count Consumer Records : " + consumerRecords.count());
      System.out.println("");
  
      Iterator&lt;ConsumerRecord&lt;String, String&gt;&gt; itr = consumerRecords.iterator();
      Map&lt;TopicPartition, OffsetAndMetadata&gt; partationOffsetMap = new HashMap&lt;&gt;(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();
  
  }
  </code></pre>
  
  <p>I just want to fetch only unread messages in kafka Consumer. Please correct me if I am wrong somewhere. And Thanks in Advance</p>