title: Spring Kafka @DltHandler annotated method is not receiving headers correctly in Non-Blocking retries implementation
body: <p>I’m trying to implement a Non-Blocking Retries with Spring Kafka. According to documentation <a href="https://docs.spring.io/spring-kafka/reference/html/#retry-topic" rel="nofollow noreferrer">here</a>, we could set up a handler method to process messages from a DLT topic, after the exhausted tries, set up in @KafkaListnener, have been done. My intend is capture some headers on DLT handler method as demonstrated in the following code:</p>
  <pre><code>@DltHandler
      fun processaDlt(
          @Payload mensagem: String,
          @Header(&quot;event&quot;) eventName: String,
          @Header(KafkaHeaders.ORIGINAL_OFFSET) offset: String,
          @Header(KafkaHeaders.EXCEPTION_FQCN) descException: String,
          @Header(KafkaHeaders.EXCEPTION_STACKTRACE) stacktrace: String,
          @Header(KafkaHeaders.EXCEPTION_MESSAGE) errorMessage: String
      ) {
  </code></pre>
  <p>But, some headers were not coming correctly, or was not coming at all. I've tried some values for these headers, like KafkaHeaders.DLT_ORIGINAL_OFFSET, KafkaHeaders.OFFSET, etc. I saw in Spring Kafka code that some headers are prefixed with the string &quot;kafka_&quot; and I saw these values on the messages that are forward to xpto-dlt topic after unsuccessful retries, <strong>but the values of some headers are truncated</strong>, as:</p>
  <blockquote>
  <p>kafka_original-offset: �, kafka_original-partition: , kafka_original-timestamp: {�<em>XR, kafka_original-timestamp-type: CreateTime, kafka_original-topic: xpto-topic, retry_topic-attempts: , retry_topic-backoff-timestamp: {�</em>�@, retry_topic-original-timestamp: {�*XR</p>
  </blockquote>
  <p>My listen method that consume the original messages are anotated as following code:</p>
  <pre><code>@KafkaListener(topics = xpto-topic, groupId = my-group-to-xpto)
  fun listen(@Payload mensagem: String,
             @Header(&quot;event&quot;) event: String,
             @Header(KafkaHeaders.OFFSET) offset: Long,
             @Header(KafkaHeaders.CONSUMER)  consumer: KafkaConsumer&lt;String, String&gt;,
             @Header(KafkaHeaders.RECEIVED_TIMESTAMP) timestamp: Long
  ) {
  </code></pre>
  <p>After all, which headers the method annotated with @DltHandler can accept? Why some values have the values truncated?</p>
  <p>Obs:</p>
  <ul>
  <li>The codes above are in Kotlin with Spring Boot</li>
  <li>I'm using Spring Boot 2.5.4-RELEASE and:</li>
  </ul>
  <blockquote>
  <p>compile 'org.springframework.kafka:spring-kafka:2.7.6'</p>
  </blockquote>
  <blockquote>
  <p>compile 'org.apache.kafka:kafka-clients:2.8.0'</p>
  </blockquote>