title: Spring Kafka: How to discard messages already retrieved by poll() after doing a seek()?
body: <p>This is a followup question to - <a href="https://stackoverflow.com/questions/39536012/reading-the-same-message-several-times-from-kafka">Reading the same message several times from Kafka</a>. If there is a better way to ask this question without posting a new question, let me know. In this post Gary mentions </p>
  
  <blockquote>
    <p>"But you will still see later messages first if they have already been retrieved so you will have to discard those too."</p>
  </blockquote>
  
  <p>Is there a clean way to discard messages already read by poll() after calling seek()? I started implementing logic to do this by saving the initial offset (in recordOffset), incrementing it on success. On failure, I call seek() and set the value of recordOffset to record.offset(). Then for every new message I check to see if the record.offset() is greater than recordOffset. If it is, I simply call acknowledge(), thereby "discarding" all the previously read messages. Here is the code - </p>
  
  <pre><code>    // in onMessage()...
      if (record.offset() &gt; recordOffset){
          acknowledgment.acknowledge();
          return;
      }
  
      try {
          processRecord(record);
          recordOffset = record.offset()+1;
          acknowledgment.acknowledge();
      } catch (Exception e) {
          recordOffset = record.offset();
          consumerSeekCallback.seek(record.topic(), record.partition(), record.offset());
      }
  </code></pre>
  
  <p>The problem with this approach is that it gets complicated with multiple partitions. Is there an easier/cleaner way?</p>
  
  <p><strong>EDIT 1</strong>
  Based on Gary's suggestion below, I tried adding an errorHandler like this -</p>
  
  <pre><code>@KafkaListener(topicPartitions =
          {@org.springframework.kafka.annotation.TopicPartition(topic = "${kafka.consumer.topic}", partitions = { "1" })},
          errorHandler = "SeekToCurrentErrorHandler")
  </code></pre>
  
  <p>Is there something wrong with this syntax as I get "Cannot resolve method 'errorHandler'"?</p>
  
  <p><strong>EDIT 2</strong>
  After Gary explained the 2 error handlers, I removed the above errorHandler and added below to the config file -</p>
  
  <pre><code>@Bean
  public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() {
      ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
      factory.setConsumerFactory(new DefaultKafkaConsumerFactory&lt;&gt;(kafkaProps()));
      factory.getContainerProperties().setAckOnError(false);
      factory.getContainerProperties().setErrorHandler(new SeekToCurrentErrorHandler());
      factory.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.MANUAL);
      return factory;
  }
  </code></pre>
  
  <p>When I start the application, I get this error now...</p>
  
  <blockquote>
    <p>java.lang.NoSuchMethodError: org.springframework.util.Assert.state(ZLjava/util/function/Supplier;)V
        at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.determineInferredType(MessagingMessageListenerAdapter.java:396)</p>
  </blockquote>
  
  <p>Here is line 396 -</p>
  
  <pre><code>Assert.state(!this.isConsumerRecordList || validParametersForBatch,
              () -&gt; String.format(stateMessage, "ConsumerRecord"));
  Assert.state(!this.isMessageList || validParametersForBatch,
              () -&gt; String.format(stateMessage, "Message&lt;?&gt;"));
  </code></pre>