title: Spring kafka Batch Listener- commit offsets manually in Batch
body: <p>I am implementing spring kafka batch listener, which reads list of messages from Kafka topic and posts the data to a REST service.
  I would like to understand the offset management in case of the REST service goes down, the offsets for the batch should not be committed and the messages should be processed for the next poll. I have read spring kafka documentation but there is confusion in understanding the difference between Listener Error Handler and Seek to current container error handlers in batch. I am using spring-boot-2.0.0.M7 version and below is my code.</p>
  
  <pre><code>Listener Config:
  
  @Bean
      KafkaListenerContainerFactory&lt;ConcurrentMessageListenerContainer&lt;String, String&gt;&gt; kafkaListenerContainerFactory() {
          ConcurrentKafkaListenerContainerFactory&lt;String, String&gt; factory = new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
          factory.setConsumerFactory(consumerFactory());
  
          factory.setConcurrency(Integer.parseInt(env.getProperty("spring.kafka.listener.concurrency")));
          // factory.getContainerProperties().setPollTimeout(3000);
          factory.getContainerProperties().setBatchErrorHandler(kafkaErrorHandler());
  
          factory.getContainerProperties().setAckMode(AckMode.BATCH);
          factory.setBatchListener(true);
          return factory;
      }
  @Bean
      public Map&lt;String, Object&gt; consumerConfigs() {
          Map&lt;String, Object&gt; propsMap = new HashMap&lt;&gt;();
          propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, env.getProperty("spring.kafka.bootstrap-servers"));
          propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,
                  env.getProperty("spring.kafka.consumer.enable-auto-commit"));
          propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,
                  env.getProperty("spring.kafka.consumer.auto-commit-interval"));
          propsMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, env.getProperty("spring.kafka.session.timeout"));
          propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
          propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
          propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, env.getProperty("spring.kafka.consumer.group-id"));
          return propsMap;
      }
  
  Listener Class:
  
  @KafkaListener(topics = "${spring.kafka.consumer.topic}", containerFactory = "kafkaListenerContainerFactory")
      public void listen(List&lt;String&gt; payloadList) throws Exception {
          if (payloadList.size() &gt; 0)
              //Post to the service
      }
  
  Kafka Error Handler:
  
  public class KafkaErrorHandler implements BatchErrorHandler {
  
      private static Logger LOGGER = LoggerFactory.getLogger(KafkaErrorHandler.class);
  
      @Override
      public void handle(Exception thrownException, ConsumerRecords&lt;?, ?&gt; data) {
          LOGGER.info("Exception occured while processing::" + thrownException.getMessage());
  
              }
  
  }
  </code></pre>
  
  <p>How to handle Kafka listener so that if something happens during processing batch of records, I wouldn't loose data.  </p>