title: How to acknowledge kafka message read by the consumer using spring integration kafka
body: <p>We are using spring-integration-kafka version 3.1.2.RELEASE and int-kafka:message-driven-channel-adapter to consume messages from the remote kafka topic. The producer sends the encrypted message and we are decrypting the actual message using deserializer. We are able to consume all the messages posted in the topic. We have used the auto commit as false. We would like to know how to commit or acknowledge the message from our service after successfully processed the message. Can someone help us how to commit the messages read from message driven channel and provide some reference implementation ?</p>
  
  <p>When we set the auto commit to true, we assume that it will commit the message after the commit interval but we would like to handle it in our service. I have come across the below example but we receive a custom object after deserialization rather spring integration message.
  so we would like to know how to implement the similar acknowledgement in the transformer so that we will not commit the message in case of any errors during the transformation. Commit the message after successful transformation.</p>
  
  <pre><code>  Acknowledgment acknowledgment = message.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class); 
    if(acknowledgment != null) { System.out.println("Acknowledgment provided");
    acknowledgment.acknowledge(); }
     }
  
  
  
  &lt;int-kafka:message-driven-channel-adapter
      id="kafkaMessageListener"
      listener-container="kafkaMessageContainer" auto-startup="true"
      phase="100" send-timeout="5000" mode="record"
      message-converter="messageConverter"
      recovery-callback="recoveryCallback" error-message-strategy="ems"
      channel="inputFromKafkaChannel" error-channel="errorChannel" /&gt;
  
  &lt;int:transformer id="transformerid"
      ref="transformerBean"
      input-channel="inputFromKafkaChannel" method="transform"
      output-channel="messageTransformer" /&gt;
  
  &lt;bean id="container1" class="org.springframework.kafka.listener.KafkaMessageListenerContainer"&gt;
      &lt;constructor-arg&gt;
          &lt;bean class="org.springframework.kafka.core.DefaultKafkaConsumerFactory"&gt;
              &lt;constructor-arg&gt;
                  &lt;map&gt;
                      &lt;entry key="bootstrap.servers" value="${spring.kafka.bootstrap-servers}" /&gt;
                      &lt;entry key="enable.auto.commit" value="false" /&gt;
                      &lt;entry key="auto.commit.interval.ms" value="100" /&gt;
                      &lt;entry key="session.timeout.ms" value="15000" /&gt;
                      &lt;entry key="group.id" value="${spring.kafka.consumer.group-id}" /&gt;
                      &lt;entry key="key.deserializer"
                          value="org.apache.kafka.common.serialization.StringDeserializer" /&gt;
                      &lt;entry key="value.deserializer"
                          value="com.test.CustomDeserializer" /&gt;
                  &lt;/map&gt;
              &lt;/constructor-arg&gt;
          &lt;/bean&gt;
      &lt;/constructor-arg&gt;
      &lt;constructor-arg&gt;
          &lt;bean class="org.springframework.kafka.listener.ContainerProperties"&gt;
              &lt;constructor-arg name="topics" value="${spring.kafka.topics}" /&gt;
          &lt;/bean&gt;
      &lt;/constructor-arg&gt;
  &lt;/bean&gt;
  </code></pre>