title: Kafka Consumer committing manually based on a condition.
body: <p>@kafkaListener consumer is commiting once a specific condition is met. Let us say a topic gets the following data from a producer
  "Message 0" at offset[0]
  "Message 1" at offset[1]</p>
  
  <p>They are received at the consumer and commited with help of acknowledgement.acknowledge()</p>
  
  <p>then the below messages come to the topic</p>
  
  <p>"Message 2" at offset[2]
  "Message 3" at offset[3]</p>
  
  <p>The consumer which is running receive the above data. Here condition fail and the above offsets are not committed. </p>
  
  <p>Even if new data comes at the topic, then also "Message 2" and "Message 3" should be picked up by any consumer from the same consumer group as they are not committed. But this is not happening,the consumer picks up a new message. </p>
  
  <p>When I restart my consumer then I get back Message2 and Message3. This should have happened while the consumers were running.</p>
  
  <p>The code is as follows -:
  KafkaConsumerConfig file</p>
  
  <pre><code>enter code here
  
  @Configuration
  @EnableKafka
  public class KafkaConsumerConfig {
      @Bean
      KafkaListenerContainerFactory&lt;ConcurrentMessageListenerContainer&lt;String, String&gt;&gt; kafkaListenerContainerFactory() {
          ConcurrentKafkaListenerContainerFactory&lt;String, String&gt; factory = new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
          factory.setConsumerFactory(consumerFactory());
          factory.setConcurrency(3);
          factory.setBatchListener(true);
  factory.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.MANUAL_IMMEDIATE);
          factory.getContainerProperties().setSyncCommits(true);
          return factory;
      }
  
      @Bean
      public ConsumerFactory&lt;String, String&gt; consumerFactory() {
          return new DefaultKafkaConsumerFactory&lt;&gt;(consumerConfigs());
      }
  
      @Bean
      public Map&lt;String, Object&gt; consumerConfigs() {
          Map&lt;String, Object&gt; propsMap = new HashMap&lt;&gt;();
          propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
          propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
          propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "100");
          propsMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000");
          propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
          propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
          propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, "group1");
          propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
          propsMap.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,"1");
          return propsMap;
      }
  
      @Bean
      public Listener listener() {
          return new Listener();
      }
  }
  
  Listner Class
  public class Listener {
      public CountDownLatch countDownLatch0 = new CountDownLatch(3);
      private Logger LOGGER = LoggerFactory.getLogger(Listener.class);
      static int count0 =0;
  
  
      @KafkaListener(topics = "abcdefghi", group = "group1", containerFactory = "kafkaListenerContainerFactory")
      public void listenPartition0(String data, @Header(KafkaHeaders.RECEIVED_PARTITION_ID) List&lt;Integer&gt; partitions,
                                   @Header(KafkaHeaders.OFFSET) List&lt;Long&gt; offsets, Acknowledgment acknowledgment) throws InterruptedException {
          count0 = count0 + 1;
          LOGGER.info("start consumer 0");
  
          LOGGER.info("received message via consumer 0='{}' with partition-offset='{}'", data, partitions + "-" + offsets);
          if (count0%2 ==0)
              acknowledgment.acknowledge();
          LOGGER.info("end of consumer 0");
  
  
      }
  </code></pre>
  
  <p>How can i achieve my desired result?</p>