Spring Boot Kafka消费者配置DefaultErrorHandler后重复打印异常栈的解决方法

核心思路 #

当前每次重试时的ERROR日志来自KafkaMessageListenerContainer的默认输出，我们可以通过调整DefaultErrorHandler的日志配置，或自定义错误处理逻辑，实现仅在最终重试失败时记录ERROR级日志。

方法1：调整DefaultErrorHandler的日志级别 #

DefaultErrorHandler支持设置重试过程中异常的日志级别，将其设为DEBUG或更低级别后，中间重试的异常不会以ERROR级别打印，仅在最终失败时通过自定义的ConsumerRecordRecoverer输出ERROR日志。

修改你的工厂代码：

public <K, V> ConcurrentKafkaListenerContainerFactory<K, V> createDlqContainerFactory(
        ConsumerFactory<K, V> consumerFactory) {
    ConcurrentKafkaListenerContainerFactory<K, V> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory);
    FixedBackOff fixedBackOff = new FixedBackOff(dlqRetryInterval, dlqMaxAttempts);
    ConsumerRecordRecoverer recovery = (record, ex) -> {
        log.error("Final retry failed for DLQ record: topic={}, partition={}, offset={}",
                record.topic(), record.partition(), record.offset(), ex);
    };
    DefaultErrorHandler errorHandler = new DefaultErrorHandler(recovery, fixedBackOff);
    // 设置重试时的日志级别为DEBUG，避免每次重试都打ERROR日志
    errorHandler.setLogLevel(Level.DEBUG);
    factory.setCommonErrorHandler(errorHandler);
    return factory;
}

方法2：直接禁用DefaultErrorHandler的默认日志输出 #

如果不需要保留任何重试过程中的日志，可直接关闭默认日志输出，仅依赖自定义恢复逻辑记录最终失败：

DefaultErrorHandler errorHandler = new DefaultErrorHandler(recovery, fixedBackOff);
// 禁用默认日志输出
errorHandler.setLogLevel(null);

方法3：自定义ErrorHandler实现细粒度控制 #

若需要更灵活的日志控制逻辑，可以继承DefaultErrorHandler，覆盖核心方法判断是否达到最大重试次数，仅在最终失败时记录日志：

public class CustomDefaultErrorHandler extends DefaultErrorHandler {

    private static final Logger log = LoggerFactory.getLogger(CustomDefaultErrorHandler.class);

    public CustomDefaultErrorHandler(ConsumerRecordRecoverer recoverer, BackOff backOff) {
        super(recoverer, backOff);
    }

    @Override
    protected void handleRemaining(Exception thrownException, List<ConsumerRecord<?, ?>> records,
                                   Consumer<?, ?> consumer, MessageListenerContainer container) {
        // 判断是否已触发终止重试的信号
        BackOffExecution backOffExecution = getBackOffExecution(records.get(0));
        if (backOffExecution.nextBackOff() == BackOffExecution.STOP) {
            log.error("Final retry failed, triggering recovery process", thrownException);
        }
        // 调用父类逻辑执行最终恢复
        super.handleRemaining(thrownException, records, consumer, container);
    }
}

在工厂中使用自定义ErrorHandler：

factory.setCommonErrorHandler(new CustomDefaultErrorHandler(recovery, fixedBackOff));

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