Spring Kafka @DltHandler 注释方法在非阻塞重试实现中未正确接收标头答案

【问题标题】：Spring Kafka @DltHandler annotated method is not receiving headers correctly in Non-Blocking retries implementationSpring Kafka @DltHandler 注释方法在非阻塞重试实现中未正确接收标头
【发布时间】：2021-11-12 16:36:46
【问题描述】：

我正在尝试使用 Spring Kafka 实现非阻塞重试。根据文档here，我们可以设置一个处理程序方法来处理来自 DLT 主题的消息，在完成了在@KafkaListnener 中设置的所有尝试之后。我的意图是在 DLT 处理程序方法上捕获一些标头，如以下代码所示：

@DltHandler
    fun processaDlt(
        @Payload mensagem: String,
        @Header("event") 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
    ) {

但是，有些标题没有正确出现，或者根本没有出现。我已经为这些标头尝试了一些值，例如 KafkaHeaders.DLT_ORIGINAL_OFFSET、KafkaHeaders.OFFSET 等。我在 Spring Kafka 代码中看到一些标头以字符串“kafka_”为前缀，我在转发到的消息中看到了这些值xpto-dlt 主题重试不成功后，但某些标头的值被截断，如：

kafka_original-offset: �, kafka_original-partition: , kafka_original-timestamp: {�XR, kafka_original-timestamp-type: CreateTime, kafka_original-topic: xpto-topic, retry_topic-attempts: , retry_topic-backoff-时间戳：{��@，retry_topic-original-timestamp：{�*XR

我使用原始消息的监听方法被注释为以下代码：

@KafkaListener(topics = xpto-topic, groupId = my-group-to-xpto)
fun listen(@Payload mensagem: String,
           @Header("event") event: String,
           @Header(KafkaHeaders.OFFSET) offset: Long,
           @Header(KafkaHeaders.CONSUMER)  consumer: KafkaConsumer<String, String>,
           @Header(KafkaHeaders.RECEIVED_TIMESTAMP) timestamp: Long
) {

毕竟，@DltHandler 注解的方法可以接受哪些标头？为什么有些值会被截断？

观察：

上面的代码是在带有 Spring Boot 的 Kotlin 中
我正在使用 Spring Boot 2.5.4-RELEASE 并且：

编译'org.springframework.kafka:spring-kafka:2.7.6'

编译'org.apache.kafka:kafka-clients:2.8.0'

【问题讨论】：

标签： spring kotlin error-handling spring-kafka dlt

【解决方案1】：

这些标头是从byte[] 转换而来的。

偏移量似乎有转换问题-当我将参数声明为long时，它返回零。

这对我来说很好......

@RetryableTopic(attempts = "1")
@KafkaListener(id = "so69229529", topics = "so69229529")
void listen(String in) {
    throw new RuntimeException();
}

@DltHandler
void handler(Message<?> msg,
        @Header(KafkaHeaders.ORIGINAL_OFFSET) byte[] offset,
        @Header(KafkaHeaders.EXCEPTION_FQCN) String descException,
        @Header(KafkaHeaders.EXCEPTION_STACKTRACE) String stacktrace,
        @Header(KafkaHeaders.EXCEPTION_MESSAGE) String errorMessage) {
    System.out.println(msg);
    System.out.println(ByteBuffer.wrap(offset).getLong());
    System.out.println(descException);
    System.out.println(stacktrace);
    System.out.println(errorMessage);
}

4
org.springframework.kafka.listener.ListenerExecutionFailedException
org.springframework.kafka.listener.ListenerExecutionFailedException: Listener failed; nested exception is org.springframework.kafka.listener.TimestampedException: Listener method 'void com.example.demo.So69229529Application.listen(java.lang.String)' threw exception; nested exception is java.lang.RuntimeException; nested exception is org.springframework.kafka.listener.ListenerExecutionFailedException: Listener method 'void com.example.demo.So69229529Application.listen(java.lang.String)' threw exception; nested exception is java.lang.RuntimeException
...
Caused by: java.lang.RuntimeException
...
Listener failed; nested exception is org.springframework.kafka.listener.TimestampedException: Listener method 'void com.example.demo.So69229529Application.listen(java.lang.String)' threw exception; nested exception is java.lang.RuntimeException; nested exception is org.springframework.kafka.listener.ListenerExecutionFailedException: Listener method 'void com.example.demo.So69229529Application.listen(java.lang.String)' threw exception; nested exception is java.lang.RuntimeException

https://github.com/spring-projects/spring-kafka/issues/1951

【讨论】：