title: How to seek a particular offset in kafka listener method?
body: <p>I am trying to seek offset from a SQL database in my kafka listener method .
  I have used registerSeekCallback method in my code but this method gets invoked when we run the consumer (or container is started) . Let's say my consumer is running and last committed offset is 20 in MySql database. I manually change the last committed offset in Mysql database to 11 but my consumer will keep reading  from 21 unless i restart my consumer(container restarted) . I am looking out for any option if i can override or seek offset in my listener method itself. Any help would be appreciated.</p>
  <pre><code>public class Listen implements ConsumerSeekAware 
  {
   @Override
      public void registerSeekCallback(ConsumerSeekCallback callback)
  
      {
  //      fetching offset from a database 
          Integer offset = offsetService.getOffset();
          callback.seek(&quot;topic-name&quot;,0,offset);
  
      }
   @KafkaListener(topics = &quot;topic-name&quot;, groupId = &quot;group&quot;)
    public void listen(ConsumerRecord record Acknowledgment acknowledgment) throws Exception 
    {
  //    processing the record 
  
        acknowledgment.acknowledge();    //manually commiting the record
  //    committing the offset to MySQL database
    }
  }
  </code></pre>
  <p>Editing with new listener method :-</p>
  <pre><code>@KafkaListener(topics = &quot;topic-name&quot;, groupId = &quot;group&quot;)
    public void listen(ConsumerRecord record Acknowledgment acknowledgment, 
    @Header(KafkaHeaders.CONSUMER) Consumer&lt;?, ?&gt; consumer)) throws Exception {
         // seeking old offset stored in database (which is 11 )
          consumer.seek(partition,offsetService.getOffset());
          log.info(&quot;record offset is {} and value is {}&quot; , record.offset(),record.value() );
          acknowledgment.acknowledge();
  }
  
  </code></pre>
  <p>In database my last committed offset is 11 and last committed offset on kafka end is 21. When i wrote a new record in kafka topic(i.e on offset 22) , my consumer triggers and processes 22 offset first then it goes back to seek offset 11 &amp; start processing from there.
  why is it consuming offset 22 first although i am seeking offset 11 ?</p>
  <p>With my above code , every time i write a new message to my kafka top it processes that record first then it seeks the offset present in my database . Is there any way i can avoid that ?</p>