title: Spring kafka : Kafka Listener- consumer.seek issue
body: <p>we are using Spring KafkaListener which acknowledges each records after it is processed to DB. If we have problems writing to DB we don't acknowledge the record so that offsets are not committed for the consumer. this works fine. Now we want to get the failed messages in next poll to retry them. we added errorhandler to our listener and invoked ConsumerAwareListenerErrorHandler and tried to do consumer.seek() for the failed message offset. Expectation is during next poll, we should received the failed messages. This is not happening. Next poll fetches only the new messages and not the failed messages Code snippet is given below.</p>
  
  <pre><code>@Service
  public class KafkaConsumer {
         @KafkaListener(topics = ("${kafka.input.stream.topic}"), containerFactory = "kafkaManualAckListenerContainerFactory", errorHandler = "listen3ErrorHandler")
        public void onMessage(ConsumerRecord&lt;Integer, String&gt; record,
              Acknowledgment acknowledgment ) throws Exception {
  
      try {
  
          msg = JaxbUtil.convertJsonStringToMsg(record.value());
          onHandList = DCMUtil.convertMsgToOnHandDTO(msg);
  
          TeradataDAO.updateData(onHandList);
          acknowledgment.acknowledge();
          recordSuccess = true;
  
          LOGGER.info("Message Saved in Teradata DB");
  
      }  catch (Exception e) {
          LOGGER.error("Error Processing On Hand Data ", e);
          recordSuccess = false;
      }
  
  }
  
    @Bean
      public ConsumerAwareListenerErrorHandler listen3ErrorHandler() throws InterruptedException {
          return (message, exception, consumer) -&gt; {
              this.listen3Exception = exception;
              MessageHeaders headers = message.getHeaders();
              consumer.seek(new org.apache.kafka.common.TopicPartition(
                            headers.get(KafkaHeaders.RECEIVED_TOPIC, String.class),
                            headers.get(KafkaHeaders.RECEIVED_PARTITION_ID, Integer.class)),
                            headers.get(KafkaHeaders.OFFSET, Long.class)); 
              return null;
          };
      }
  }
  
      Container  Class
  
      @Bean
  public Map&lt;Object,Object&gt; consumerConfigs() {
      Map&lt;Object,Object&gt;  props = new HashMap&lt;Object,Object&gt; ();
      props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
              localhost:9092);
      props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
              StringDeserializer.class);
      props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
              StringDeserializer.class);
      props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
      props.put(ConsumerConfig.GROUP_ID_CONFIG, "example-1");
      props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
      return props;
  }
  
  @SuppressWarnings({ "rawtypes", "unchecked" })
  @Bean
  public ConsumerFactory consumerFactory() {
      return new DefaultKafkaConsumerFactory(consumerConfigs());
  }
  
   @SuppressWarnings("unchecked")
  @Bean
  KafkaListenerContainerFactory&lt;ConcurrentMessageListenerContainer&lt;Integer, String&gt;&gt;
  kafkaManualAckListenerContainerFactory() {
      ConcurrentKafkaListenerContainerFactory&lt;Integer, String&gt; factory =
              new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
      factory.setConsumerFactory(consumerFactory());
      factory.getContainerProperties().setAckMode(AckMode.MANUAL);
      return factory;
  }
  </code></pre>