title: spring-kafka consumer batch error handling with spring boot version 2.3.7
body: <p>I am trying to perform the spring kafka batch process error handling. First of all i have few questions.</p>
  <ol>
  <li><p>what is difference between listener and container error handlers and what errors comes into these two categories ?</p>
  </li>
  <li><p>Could you please help some samples on this to understand better ?</p>
  </li>
  </ol>
  <p>Here is our design:</p>
  <ol>
  <li>Poll every certain interval</li>
  <li>consume messages in a batch mode</li>
  <li>push to local cache (application cache) based on key (to avoid duplicate events)</li>
  <li>push all values one by one to another topic once batch process done.</li>
  <li>clear the the cache once the operation 3 done and acknowledge the offsets manually.</li>
  </ol>
  <p>Here is my plan to have error handling:</p>
  <pre><code>public ConcurrentKafkaListenerContainerFactory&lt;String, String&gt; myListenerPartitionContainerFactory(String groupId) {
          ConcurrentKafkaListenerContainerFactory&lt;String, String&gt; factory = new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
          factory.setConsumerFactory(consumerFactory(groupId));
          factory.setConcurrency(partionCount);
          factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
          factory.getContainerProperties().setIdleBetweenPolls(pollInterval);
          factory.setBatchListener(true);
  
          return factory;
      }
  
      @Bean
      public ConcurrentKafkaListenerContainerFactory&lt;String, String&gt; myPartitionsListenerContainerFactory() 
      {
          return myListenerPartitionContainerFactory(groupIdPO);
      }
  
  
  @Bean
  public RecoveringBatchErrorHandler(KafkaTemplate&lt;String, String&gt; errorKafkaTemplate) {
      DeadLetterPublishingRecoverer recoverer =
              new DeadLetterPublishingRecoverer(errorKakfaTemplate);
      RecoveringBatchErrorHandler errorHandler =
              new RecoveringBatchErrorHandler(recoverer, new FixedBackOff(2L, 5000)); // push error event to the error topic
  }
  
  
  @KafkaListener(id = &quot;mylistener&quot;, topics = &quot;someTopic&quot;, containerFactory = &quot;myPartitionsListenerContainerFactory&quot;))
  public void listen(List&lt;ConsumerRecord&lt;String, String&gt;&gt; records, @Header(KafkaHeaders.MESSAGE_KEY) String key, Acknowledgement ack) {
      Map hashmap = new Hashmap&lt;&gt;();
      records.forEach(record -&gt; {
          try {
              //key will be formed based on the input record - it will be id.
              hashmap.put(key, record);  
          }
          catch (Exception e) {
              throw new BatchListenerFailedException(&quot;Failed to process&quot;, record);
          }
           
      });
      // Once success each messages to another topic.
      try {
        hashmap.forEach( (key,value) -&gt; {  push to another topic })
        hashmap.clear();
        ack.acknowledge();
      } catch(Exception ex) {
          //handle producer exceptions
      }
  }
  </code></pre>
  <p>is the direction good or any improvements needs to be done? And also what type of container and listener handlers need to be implemented?</p>
  <p>@Gary Russell.. could you please help on this ?</p>