title: How to use Spring Kafka's Acknowledgement.acknowledge() method 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 as mentioned here <a href="https://docs.spring.io/spring-kafka/reference/html/_reference.html#committing-offsets" rel="noreferrer">https://docs.spring.io/spring-kafka/reference/html/_reference.html#committing-offsets</a>. Mine is spring-boot application. If I am not using manual commit process than my code is working fine. But when I use 
  Acknowledgement.acknowledge() for manual commit it shows error related to bean. Also If I am not using manual commit properly please suggest me the right way to do it.</p>
  
  <p>Error message:</p>
  
  <pre><code>***************************
  APPLICATION FAILED TO START
  ***************************
  
  Description:
  
  Field ack in Receiver required a bean of type 'org.springframework.kafka.support.Acknowledgment' that could not be found.
  
  
  Action:
  
  Consider defining a bean of type 'org.springframework.kafka.support.Acknowledgment' in your configuration.
  </code></pre>
  
  <p>I googled this error I found that I need to add @Component but that is already there in my consumer code.</p>
  
  <p>My consumer code looks like this: Receiver.java</p>
  
  <pre><code>import java.util.concurrent.CountDownLatch;
  import org.apache.kafka.clients.consumer.ConsumerRecord;
  import org.springframework.beans.factory.annotation.Autowired;
  import org.springframework.kafka.annotation.KafkaListener;
  import org.springframework.kafka.support.Acknowledgment;
  import org.springframework.stereotype.Component;
  
  @Component
  public class Receiver {
  
      @Autowired
      public Acknowledgment ack;
  
      private CountDownLatch latch = new CountDownLatch(1);
  
      @KafkaListener(topics = "${kafka.topic.TestTopic}")
      public void receive(ConsumerRecord&lt;?, ?&gt; consumerRecord){
              System.out.println(consumerRecord.value());
              latch.countDown();
              ack.acknowledge();
      }
  }
  </code></pre>
  
  <p>My producer code looks like this: Sender.java</p>
  
  <pre><code>import java.util.Map;
  import org.springframework.beans.factory.annotation.Autowired;
  import org.springframework.kafka.core.KafkaTemplate;
  import org.springframework.stereotype.Component;
  
  @Component
  public class Sender {
  
      @Autowired
      private KafkaTemplate&lt;String, Map&lt;String, Object&gt;&gt; kafkaTemplate;
  
      public void send(Map&lt;String, Object&gt; map){
              kafkaTemplate.send("TestTopic", map);
  
      }
  
  }
  </code></pre>
  
  <p><strong>EDIT 1:</strong></p>
  
  <p>My new consumer code looks like this: Receiver.java</p>
  
  <pre><code>import java.util.concurrent.CountDownLatch;
  import org.apache.kafka.clients.consumer.ConsumerRecord;
  import org.springframework.kafka.annotation.KafkaListener;
  import org.springframework.kafka.support.Acknowledgment;
  import org.springframework.stereotype.Component;
  
  @Component
  public class Receiver {
  
      private CountDownLatch latch = new CountDownLatch(1);
  
      @KafkaListener(topics = "${kafka.topic.TestTopic}", containerFactory = "kafkaManualAckListenerContainerFactory")
      public void receive(ConsumerRecord&lt;?, ?&gt; consumerRecord, Acknowledgment ack){
              System.out.println(consumerRecord.value());
              latch.countDown();
              ack.acknowledge();
      }
  }
  </code></pre>
  
  <p>I changed my configuration class also:</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.ConsumerFactory;
  import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
  
  @Configuration
  @EnableKafka
  public class ReceiverConfig {
  
      @Value("${kafka.bootstrap-servers}")
      private String bootstrapServers;
  
      @Value("${spring.kafka.consumer.group-id}")
      private String consumerGroupId;
  
      @Bean
      public Map&lt;String, Object&gt; consumerConfigs() throws SendGridException {
              Map&lt;String, Object&gt; props = new HashMap&lt;&gt;();
              props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
              props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
              props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
              props.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroupId);
              return props;
  
      }
  
      @Bean
      public ConsumerFactory&lt;String, String&gt; consumerFactory(){
          return new DefaultKafkaConsumerFactory&lt;&gt;(consumerConfigs());
      }
  
      /*@Bean
      public ConcurrentKafkaListenerContainerFactory&lt;String, String&gt; kafkaListenerContainerFactory(){
          ConcurrentKafkaListenerContainerFactory&lt;String, String&gt; factory = new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
          factory.setConsumerFactory(consumerFactory());
  
          return factory;
      }*/
  
      @Bean
      public ConcurrentKafkaListenerContainerFactory&lt;String, String&gt; kafkaManualAckListenerContainerFactory(){
          ConcurrentKafkaListenerContainerFactory&lt;String, String&gt; factory = new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
          factory.setConsumerFactory(consumerFactory());
          return factory;
      }
  
      @Bean
      public Receiver receiver() {
          return new Receiver();
      }
  }
  </code></pre>
  
  <p>After adding containerFactory = "kafkaManualAckListenerContainerFactory" to my receive() method I am getting the below error.</p>
  
  <pre><code>***************************
  APPLICATION FAILED TO START
  ***************************
  
  Description:
  
  Parameter 1 of method kafkaListenerContainerFactory in org.springframework.boot.autoconfigure.kafka.KafkaAnnotationDrivenConfiguration required a bean of type 'org.springframework.kafka.core.ConsumerFactory' that could not be found.
      - Bean method 'kafkaConsumerFactory' in 'KafkaAutoConfiguration' not loaded because @ConditionalOnMissingBean (types: org.springframework.kafka.core.ConsumerFactory; SearchStrategy: all) found bean 'consumerFactory'
  
  
  Action:
  
  Consider revisiting the conditions above or defining a bean of type 'org.springframework.kafka.core.ConsumerFactory' in your configuration.
  </code></pre>