title: Spring Kafka Consumer with database
body: <p>How can I  execute the below in a transaction. My requirement is message offset should not be committed to Kafka if the  DB calls fails .Kafka consumer configuration is  here <a href="https://pastebin.com/kq5S9Jrx" rel="nofollow noreferrer">https://pastebin.com/kq5S9Jrx</a></p>
  <pre><code>@KafkaListener(topics = &quot;${general.topic.name}&quot; , groupId = &quot;${general.topic.group.id}&quot; )
      public void consume(String message,@Header(KafkaHeaders.ACKNOWLEDGMENT) Acknowledgment ack) 
      {
          logger.debug(String.format(&quot;Message recieved -&gt; %s&quot;, message));
          
          // start transaction
          dbservice.validateMessage(message);
          
          dbservice.saveInDB(message);
          ack.acknowledge();
          // end transaction
  }
  </code></pre>