title: How to acknowledge current offset in spring kafka for manual commit
body: <p>I am using Spring Kafka first time and I am not able to use Acknowledgement.acknowledge() method for manual commit in my consumer code. please let me know if anything missing in my consumer configuration or listener code. or else is there other way to handle acknowledge offset based on condition.
      Here i'm looking solution like if the offset is not committed/ acknowledge manually, it should pick same message/offset by consumer. </p>
  
  <p><strong>Configuration</strong></p>
  
  <pre><code>import java.util.HashMap;
  import java.util.Map;
  
  import org.apache.kafka.clients.consumer.ConsumerConfig;
  import org.apache.kafka.common.serialization.StringDeserializer;
  import org.springframework.beans.factory.annotation.Value;
  import org.springframework.context.annotation.Bean;
  import org.springframework.context.annotation.Configuration;
  import org.springframework.kafka.annotation.EnableKafka;
  import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
  import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
  import org.springframework.kafka.listener.AbstractMessageListenerContainer.AckMode;
  
  @EnableKafka
  @Configuration
  public class ConsumerConfig {
  
      @Value(value = "${kafka.bootstrapAddress}")
      private String bootstrapAddress;
  
      @Value(value = "${kafka.groupId}")
      private String groupId;
  
      @Bean
      public ConcurrentKafkaListenerContainerFactory&lt;String, String&gt; containerFactory() {
          Map&lt;String, Object&gt; props = new HashMap&lt;String, Object&gt;();
          props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
          props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
          props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
          props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 100);
          props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                  StringDeserializer.class);
          props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                  StringDeserializer.class);
          props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
          ConcurrentKafkaListenerContainerFactory&lt;String, String&gt; factory = new ConcurrentKafkaListenerContainerFactory&lt;String, String&gt;();
          factory.setConsumerFactory(new DefaultKafkaConsumerFactory&lt;String, String&gt;(
                  props));
          factory.getContainerProperties().setAckMode(AckMode.MANUAL_IMMEDIATE);
          factory.getContainerProperties().setSyncCommits(true);
          return factory;
      }
  }
  </code></pre>
  
  <p><strong>Listener</strong></p>
  
  <pre><code>private static int value = 1;
  
  @KafkaListener(id = "baz", topics = "${message.topic.name}", containerFactory = "containerFactory")
  public void listenPEN_RE(@Payload String message,
          @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
          @Header(KafkaHeaders.OFFSET) int offsets,
          Acknowledgment acknowledgment) {
  
      if (value%2==0){
          acknowledgment.acknowledge();
      }
      value++;
  }
  </code></pre>