title: How can i capture record key and value when there is a DeserializationException while consuming a message from kafka topic?
body: <p>I'm using spring boot 2.1.7.RELEASE and spring-kafka 2.2.8.RELEASE.And I'm using @KafkaListener annotation to create a consumer and I'm using all default settings for the consumer.And I'm using below configuration as specified in the Spring-Kafka documentation.</p>
  <pre><code>// other props
      props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer2.class);
      props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer2.class);
      props.put(ErrorHandlingDeserializer.KEY_DESERIALIZER_CLASS, StringDeserializer.class);
      props.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, AvroDeserializer.class.getName());
      return new DefaultKafkaConsumerFactory&lt;&gt;(props);
  </code></pre>
  <p>Now, I've implemented my custom SeekToCurrentErrorHandler by extending SeekToCurrentErrorHandler as per the below thread but the record value is coming as null and the record key is not in a readable format. Please suggest me how can i get the record key and value?</p>
  <p><a href="https://stackoverflow.com/questions/60553310/how-to-capture-the-exception-and-message-key-when-using-errorhandlingdeserialize">How to capture the exception and message key when using ErrorHandlingDeserializer2 to handle exceptions during deserialization</a></p>
  <p>Here is my custom SeekToCurrentErrorHandler code</p>
  <pre><code>@Component
  public class MySeekToCurrentErrorHandler extends SeekToCurrentErrorHandler {
  
      private final MyDeadLetterRecoverer deadLetterRecoverer;
  
      @Autowired
      public MySeekToCurrentErrorHandler(MyDeadLetterRecoverer deadLetterRecoverer) {
          super(-1);
          this.deadLetterRecoverer = deadLetterRecoverer;
      }
  
      @Override
      public void handle(Exception thrownException, List&lt;ConsumerRecord&lt;?, ?&gt;&gt; data, Consumer&lt;?, ?&gt; consumer, MessageListenerContainer container) {
          if (thrownException instanceof DeserializationException) {
              //Improve to support multiple records
              DeserializationException deserializationException = (DeserializationException) thrownException;
              deadLetterRecoverer.accept(data.get(0), deserializationException);
              
              ConsumerRecord&lt;?, ?&gt;. consumerRecord = data.get(0);
              sout(consumerRecord.key());
              sout(consumerRecord.value());
          } else {
              //Calling super method to let the 'SeekToCurrentErrorHandler' do what it is actually designed for
              super.handle(thrownException, data, consumer, container);
          }
      }
  }
  </code></pre>