启用Dead Letter Queue后无错误日志记录的技术问题咨询

处理入站消息发生错误时，失败消息会被移至Dead Letter Queue（死信队列），但导致消息处理失败的异常未被记录。

使用版本信息 #

• Spring Cloud Stream：4.0.1
• Binder：Kafka

配置详情 #

spring:
  cloud:
    function:
       definition: preProcessingMessageAudit|abcProcessing|postProcessingMessageAudit
    stream:
     function:
        bindings:
             preProcessingMessageAudit|abcProcessing|postProcessingMessageAudit-in-0: input
             preProcessingMessageAudit|abcProcessing|postProcessingMessageAudit-out-0: output
     bindings:
        input:
           destination: ABCIncomingChannel
           consumer:
            concurrency: 4
           group: mpa
        output:
           destination: ABCOutgoingChannel
           producer:
            partitionCount: 4
     kafka:
        binder:
          kafka-properties:
            bootstrap-servers: localhost:9092
          replicationFactor : 1 
          autoAddPartitions : true
        bindings:
           input:
             consumer:
                enableDlq: true
                dlqName: abc-dead-letter

1. 开启DLQ异常内容转发 #

在Kafka消费者绑定配置中添加dlq-emit-error-record-content并设为true，该配置会将异常信息附加到DLQ消息中，同时框架会自动记录错误日志：

spring:
  cloud:
    stream:
      kafka:
        bindings:
           input:
             consumer:
                enableDlq: true
                dlqName: abc-dead-letter
                dlq-emit-error-record-content: true

2. 调整日志级别 #

确保org.springframework.cloud.stream.binder.kafka包的日志级别为ERROR或WARN，在配置文件中添加：

logging:
  level:
    org.springframework.cloud.stream.binder.kafka: ERROR

3. 自定义异常记录逻辑（可选） #

如果需要更定制化的日志输出，可以通过自定义错误处理逻辑捕获异常并记录：

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.context.annotation.Bean;
import java.util.function.Consumer;

public class CustomErrorHandler {
    private static final Logger log = LoggerFactory.getLogger(CustomErrorHandler.class);

    @Bean
    public Consumer<Message<?>> messageErrorHandler() {
        return message -> {
            Throwable exception = (Throwable) message.getHeaders().get(MessageHeaders.ERROR_CHANNEL_EXCEPTION);
            log.error("消息处理失败，消息内容：{}", message.getPayload(), exception);
        };
    }
}

之后可结合Spring Cloud Stream的错误通道机制，将该处理逻辑接入到消息错误流转链路中。

内容的提问来源于stack exchange，提问作者HashDhi