Spring Cloud Stream Kafka仅针对特定异常发送消息至DLQ问题排查

使用Spring Cloud Stream Kafka binder处理消息时，需求是仅将自定义异常UserException对应的消息发送至DLQ，而非所有RuntimeException。配置了ListenerContainerCustomizer等Bean后，当前代码仍会把所有异常消息转发至DLQ，同时尝试跳过消息头中的异常堆栈信息也未生效。

相关代码如下：

@Bean
public ListenerContainerCustomizer<AbstractMessageListenerContainer<byte[], byte[]>> customizer(DefaultErrorHandler errorHandler) {
    return (container, dest, group) -> container.setCommonErrorHandler(errorHandler);
}
@Bean
public DefaultErrorHandler errorHandler(DeadLetterPublishingRecoverer recovered) {
    recovered.excludeHeader(DeadLetterPublishingRecoverer.HeaderNames.HeadersToAdd.EX_STACKTRACE);
    recovered.addNotRetryableExceptions(UserException.class);
    return new DefaultErrorHandler(recovered, new FixedBackOff(0L, 1L));
}
@Bean
public DeadLetterPublishingRecoverer publisher(KafkaTemplate<Object, SpecificRecord> kafkaOperations) {
    return new DeadLetterPublishingRecoverer(
            kafkaOperations,
            (consumerRecord, exception) -> {
                return new TopicPartition(dlqTopicName, 0);
            }
    );
}

你的配置存在两个核心问题，对应修改方式如下：

1. 异常控制逻辑放错了位置
   addNotRetryableExceptions是DefaultErrorHandler的方法，不是DeadLetterPublishingRecoverer的。恢复器只负责转发消息到DLQ，而哪些异常需要走转发流程，是由DefaultErrorHandler来控制的。

修改方法：把异常筛选逻辑移到DefaultErrorHandler中，明确指定仅对UserException执行转发逻辑，其他异常直接跳过：

@Bean
public DefaultErrorHandler errorHandler(DeadLetterPublishingRecoverer recovered) {
    DefaultErrorHandler errorHandler = new DefaultErrorHandler(recovered, new FixedBackOff(0L, 1L));
    // 仅对UserException触发重试（这里重试次数设为0，直接进入DLQ流程）
    errorHandler.retryOn(UserException.class);
    // 排除所有其他RuntimeException，不触发重试和DLQ转发
    errorHandler.notRetryOn(RuntimeException.class);
    return errorHandler;
}

2. 排除堆栈头的方式错误
   直接调用recovered.excludeHeader()不会生效，需要在创建DeadLetterPublishingRecoverer时，通过HeaderConfig传入排除规则：

@Bean
public DeadLetterPublishingRecoverer publisher(KafkaTemplate<Object, SpecificRecord> kafkaOperations) {
    // 配置要排除的消息头
    DeadLetterPublishingRecoverer.HeaderConfig headerConfig = new DeadLetterPublishingRecoverer.HeaderConfig()
            .excludeHeaders(DeadLetterPublishingRecoverer.HeaderNames.HeadersToAdd.EX_STACKTRACE);
    return new DeadLetterPublishingRecoverer(
            kafkaOperations,
            (consumerRecord, exception) -> new TopicPartition(dlqTopicName, 0),
            headerConfig // 传入HeaderConfig参数
    );
}

3. 额外注意事项
   如果业务代码中存在异常包装（比如UserException被包装成了其他RuntimeException抛出），需要给DefaultErrorHandler添加异常解析逻辑，确保能识别到原始的UserException：

errorHandler.setExceptionHandler((record, exception) -> {
    // 递归获取根异常
    Throwable rootCause = org.springframework.util.ExceptionUtils.getRootCause(exception);
    return rootCause != null ? rootCause : exception;
});

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