title: I want to enable Kafka errorhandling by annotation
body: <p>I tried to enable spring-kafka errorhandling by annotation.</p>
  
  <p>So I created a bean <code>kafkaListenerErrorHandler</code>.</p>
  
  <pre><code>@Bean
  public KafkaListenerErrorHandler kafkaListenerErrorHandler(ErrorHandler errorHandler) {
      KafkaListenerErrorHandler kafkaListenerErrorHandler = (message, exception) -&gt; {
          MessageHeaders headers = message.getHeaders();
          ConsumerRecord&lt;String, EventV1&gt; consumerRecord = new ConsumerRecord&lt;&gt;(
                  headers.get(KafkaHeaders.RECEIVED_TOPIC, String.class),
                  headers.get(KafkaHeaders.RECEIVED_PARTITION_ID, Integer.class),
                  headers.get(KafkaHeaders.OFFSET, Long.class),
                  headers.get(KafkaHeaders.RECEIVED_MESSAGE_KEY, String.class),
                  (EventV1) message.getPayload());
          errorHandler.handle(exception, Collections.singletonList(consumerRecord), headers.get(KafkaHeaders.CONSUMER, Consumer.class), messageListenerContainer);
          return null;
      };
  
      return kafkaListenerErrorHandler;
  }
  </code></pre>
  
  <p>and wired it with the <code>KafkaListener</code> by</p>
  
  <pre><code>@Autowired(required = false)
  MessageListenerContainer messageListenerContainer;
  
  @Autowired
  private UserBO userBO;
  
  @Bean
  ErrorHandler errorHandler(KafkaTemplate kafkaTemplate) {
      return new SeekToCurrentErrorHandler(new DeadLetterPublishingRecoverer(kafkaTemplate), new FixedBackOff(0, 2));
  }
  
  @KafkaListener(topics = "#{'${kafka.listener.topics}'.split(',')}", errorHandler = "kafkaListenerErrorHandler")
  public void listen(ConsumerRecord&lt;String, EventV1&gt; message) {
  
      log.info("Received Messasge: {}", message);
      try {
          EventV1 value = message.value();
          userBO.getUserById(value.getCustomerId());
      } catch (Exception e) {
          throw new ListenerExecutionFailedException(e.getLocalizedMessage(), e);
      }
  }
  </code></pre>
  
  <p>Even if it seems to work, the code seems ugly. The <code>kafkaListenerErrorHandler</code> constructs a new ConsumerRecord to forward it to the ErrorHandler and the "messageListenerContainer" is null, because I didn't find out how to get it into my context.</p>
  
  <p>There should or must be a more elegant way to connect the ErrorHandler with the KafkaListenerErrorHandler</p>
  
  <p>I also added a deserialization error handler to my setup.</p>
  
  <p><code>spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.ErrorHandlingDeserializer2
  spring.kafka.consumer.properties.spring.deserializer.value.delegate.class=org.springframework.kafka.support.serializer.JsonDeserializer
  spring.kafka.consumer.properties.spring.json.trusted.packages=*
  </code>
  Thanks for any advice.</p>