如何在Spring Kafka非阻塞重试机制下记录@KafkaListener抛出的异常

核心原因 #

Spring Kafka 2.7 非阻塞重试机制默认不会打印@KafkaListener业务方法抛出的原始异常，仅输出seek相关提示，因为默认异常处理链路未包含原始异常的日志落盘逻辑。

配置调整方法 #

方案1：给监听器容器工厂添加异常日志处理器 #

在原有的kafkaListenerContainerFactory配置中新增CommonErrorHandler，先打印原始异常再透传给非阻塞重试链路，注意需要关闭默认的阻塞重试避免冲突：

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.util.backoff.FixedBackOff;

// 省略其他包引用
private static final Logger log = LoggerFactory.getLogger(YourConfigClass.class);

@Bean
public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, Object> factory
        = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    // 新增异常日志处理逻辑
    factory.setCommonErrorHandler(new DefaultErrorHandler((record, exception) -> {
        // 打印完整原始异常栈
        log.error("Kafka消息消费失败，topic={}, partition={}, offset={}, 消息key={}",
            record.topic(), record.partition(), record.offset(), record.key(), exception);
        // 不需要手动处理重试/死信，非阻塞重试机制会自动接管后续流程
    }, new FixedBackOff(0, 0))); // FixedBackOff设置为0次阻塞重试，避免和非阻塞重试冲突
    return factory;
}

方案2：在RetryTopic配置中添加重试监听器 #

如果需要更精准的和非阻塞重试逻辑绑定，也可以直接给RetryTopicConfiguration添加重试监听器，在每次重试触发时记录异常：

import org.springframework.kafka.retrytopic.RetryListener;
import org.apache.kafka.clients.consumer.ConsumerRecord;

// 省略其他包引用
@Bean
public RetryTopicConfiguration retryTopicConfiguration(KafkaTemplate<String, Object> template) {
    List<Class<? extends Throwable>> throwableList = Arrays.asList(IllegalArgumentException.class,
        IllegalAccessException.class);
    return RetryTopicConfigurationBuilder
        .newInstance()
        .dltHandlerMethod(XYZ.class, "xyz")
        .exponentialBackoff(delayMs, backoffMultiplier, maxIntervalInMs)
        .maxAttempts(retryAttempt)
        .notRetryOn(throwableList)
        .doNotAutoCreateRetryTopics()
        .listenerFactory(kafkaListenerContainerFactory())
        .setTopicSuffixingStrategy(TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE)
        // 新增重试监听器，每次消费抛出异常触发重试时都会执行
        .retryListener(new RetryListener() {
            @Override
            public void onRetry(ConsumerRecord<?, ?> record, Exception exception, int attempt) {
                log.error("Kafka消息第{}次重试消费失败，topic={}, offset={}",
                    attempt, record.topic(), record.offset(), exception);
            }

            @Override
            public void onDltProcessingFailure(ConsumerRecord<?, ?> record, Exception exception) {
                log.error("Kafka消息进入死信队列处理失败，topic={}, offset={}",
                    record.topic(), record.offset(), exception);
            }

            @Override
            public void onSuccess(ConsumerRecord<?, ?> record, int attempt) {
                // 可选：记录重试成功的日志
                log.info("Kafka消息第{}次重试消费成功，topic={}, offset={}",
                    attempt, record.topic(), record.offset());
            }
        })
        .create(template);
}

注意事项 #

• 两种方案二选一即可，若同时配置，方案1的日志会先触发
• 确保日志框架的error级别已经开启，否则异常日志无法正常输出
• 如果你配置了notRetryOn的异常类型，两类异常都会被上述日志逻辑捕获，可根据业务需要在日志打印前添加异常类型判断做区分

内容的提问来源于stack exchange，提问作者Pat's