title: Kafka manual commit based on conditions
body: <p>I have a Kafka Sink Task which is listening to a Kafka topic where I want the message to get processed and persisted in DB first and then want to manually commit the offset just to ensure there are no loss of messages that have come to a topic but not persisted in DB.</p>
  <p>To check if the message has been persisted, I am using a custom event with ApplicationEventPublisher in SpringBoot app which is publishing that custom event upon successful commit in DB.</p>
  <p>I am having an @EventListener listening to that custom event which has to notify @KafkaListener that the message can be acknowledged now.</p>
  <p>Also after fetching the records from Kafka, if the processing and persistence is successful then only I want to manually commit the offset else it should read from the same offset again.</p>
  <p>Below is the set of things that I have configured:-</p>
  <ul>
  <li>enable.auto.commit = false</li>
  <li>AckMode has been set to MANUAL</li>
  </ul>
  <blockquote>
  <pre><code>@KafkaListener(containerFactory = &quot;kafkaProvider&quot;,
          topics = &quot;test&quot;)
  public void receive(@Payload V payload,
                      @Header(KafkaHeaders.RECEIVED_PARTITION_ID) List&lt;Integer&gt; partitions,
                      @Header(value = KafkaHeaders.RECEIVED_MESSAGE_KEY, required = false) K key,
                      @Header(KafkaHeaders.OFFSET) int offsets,
                      Acknowledgment acknowledgment) {
      
    if(message is persisted){
  
      // This is added for manually committing the message
      acknowledgement.acknowledge();
      
      // Send the data to the provided topic with the provided key and no partition.
      kafkaTemplate.send(topic, key, value);
     } 
  }
  
  @EventListener
  private boolean checkMessagePersisted(MessageAcknowledgeEvent event) {
      if(event.isPersisted()){
          return true;
      }else{
          logger.error();
          return false;
      }
  }
  </code></pre>
  </blockquote>
  <p>Here there are 2 suggestions I need:-</p>
  <ol>
  <li>If I use acknowledgement.acknowledge() to manually commit the message, do I still have to use kafkatemplate.send() method. If not, then how will it know which key and data are required to send to topic</li>
  <li>Any best practices for sharing the flag value from @EventListener to @KafkaListener so that the message can be committed now only if the persistence flag check is true</li>
  </ol>