Spring Kafka消费者持续输出Consumer exception，如何获取真实异常堆栈？

我正在使用配置了Spring Kafka的Spring Boot应用从Kafka读取消息，版本信息如下：Spring Boot 3.2.3、Spring Kafka 3.1.1。

我的消费者在应用日志中持续抛出如下错误，且该ERROR级别的“Consumer exception”信息每秒都会出现大量重复：

Logging Level: ERROR
Message text: Consumer exception
Class name: org.springframework.kafka.listener.KafkaMessageListenerContainer

我已为消费者配置了如下错误处理机制：

1. 自定义ConsumerAwareListenerErrorHandler：

@KafkaListener(id = "test-1", topics = "my-test-topic", groupId = "my-test-consumer-group", errorHandler = "kafkaEventErrorHandler")

对应的处理器实现：

public class KafkaEventErrorHandler<T> implements ConsumerAwareListenerErrorHandler {

  @Override
  public Object handleError(Message<?> message, ListenerExecutionFailedException e,
      Consumer<?, ?> consumer) {
    log.error("Message Headers : {}", message.getHeaders());
    log.trace("Message Payload : {}", message.getPayload());
    log.error("Exception StackTrace : {}", ExceptionUtils.getStackTrace(e));
    log.info("Consumer Metrics: {}", consumer.metrics().toString());
    return null;
  }
}

2. 配置了DefaultErrorHandler Bean：

@Bean
public DefaultErrorHandler errorHandler() {
  BackOff fixedBackOff = new FixedBackOff(500, 3);
  DefaultErrorHandler errorHandler = new DefaultErrorHandler((consumerRecord, backoff) -> {
  }, fixedBackOff);
  errorHandler.addRetryableExceptions(SocketTimeoutException.class);
  errorHandler.addNotRetryableExceptions(NullPointerException.class);
  return errorHandler;
}

但异常仍未被上述处理器捕获，仅输出“Consumer exception”。我想了解：

1. 如何通过现有错误处理器提取并打印KafkaMessageListenerContainer抛出的真实异常堆栈？
2. 若无法实现，有哪些替代方案可以获取真实异常信息？

一、通过现有处理器捕获并打印真实异常 #

你看到的KafkaMessageListenerContainer输出的“Consumer exception”，大概率是消费者线程在拉取消息、提交偏移量等底层操作时抛出的异常——这类异常不会进入ConsumerAwareListenerErrorHandler（它仅处理@KafkaListener方法执行阶段的异常），且你的DefaultErrorHandler配置未覆盖底层异常场景。

1. 调整DefaultErrorHandler配置，覆盖底层异常处理 #

修改DefaultErrorHandler Bean，添加自定义恢复逻辑并确保打印完整异常堆栈，同时开启对消费者底层异常的处理：

@Bean
public DefaultErrorHandler errorHandler() {
    BackOff fixedBackOff = new FixedBackOff(500, 3);
    // 自定义失败处理器，打印完整异常信息
    ConsumerRecordRecoverer recoverer = (consumerRecord, exception) -> {
        log.error("处理Kafka消息失败，详情：topic={}, partition={}, offset={}",
                consumerRecord.topic(), consumerRecord.partition(), consumerRecord.offset(),
                exception); // 传入exception，日志框架会自动打印完整堆栈
    };
    DefaultErrorHandler errorHandler = new DefaultErrorHandler(recoverer, fixedBackOff);
    // 添加需要重试的底层异常（如网络类异常）
    errorHandler.addRetryableExceptions(SocketTimeoutException.class, IOException.class);
    // 自定义重试判断逻辑（根据实际需求调整）
    errorHandler.setRetryableExceptions((ex) -> ex instanceof SocketTimeoutException);
    // 开启对消费者底层异常的处理
    errorHandler.setAckAfterHandle(false);
    return errorHandler;
}

2. 调整日志框架配置，强制打印异常堆栈 #

KafkaMessageListenerContainer的默认ERROR日志可能仅输出消息文本，未打印堆栈。以Logback为例，在logback-spring.xml中添加如下配置：

<logger name="org.springframework.kafka.listener.KafkaMessageListenerContainer" level="ERROR">
    <appender-ref ref="CONSOLE"/>
    <appender-ref ref="FILE"/>
    <encoder>
        <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n%ex</pattern>
    </encoder>
</logger>

二、替代方案获取真实异常信息 #

如果上述调整仍无法捕获异常，可通过以下方式获取真实异常：

1. 自定义消费者拦截器 #

通过ConsumerInterceptor捕获消费者拉取、提交阶段的异常：

public class LoggingConsumerInterceptor<K, V> implements ConsumerInterceptor<K, V> {

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

    @Override
    public ConsumerRecords<K, V> onConsume(ConsumerRecords<K, V> records) {
        return records;
    }

    @Override
    public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {
        // 捕获提交偏移量时的异常
    }

    @Override
    public void close() {}

    @Override
    public void configure(Map<String, ?> configs) {}
}

然后在消费者工厂配置中添加拦截器：

@Bean
public ConsumerFactory<String, Object> consumerFactory() {
    Map<String, Object> configProps = new HashMap<>();
    configProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "你的Kafka地址");
    // 其他消费者配置...
    configProps.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, LoggingConsumerInterceptor.class.getName());
    return new DefaultKafkaConsumerFactory<>(configProps);
}

2. 开启DEBUG级日志排查 #

临时将org.springframework.kafka和org.apache.kafka的日志级别调整为DEBUG，查看消费者底层的完整操作日志（包括异常堆栈）：

<!-- Logback配置示例 -->
<logger name="org.springframework.kafka" level="DEBUG"/>
<logger name="org.apache.kafka" level="DEBUG"/>

注意：DEBUG日志量较大，排查完成后需改回原级别。

3. 自定义容器异常监听器 #

通过自定义KafkaListenerContainerFactory，添加容器级别的异常监听器：

@Bean
public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory(
        ConsumerFactory<String, Object> consumerFactory, DefaultErrorHandler errorHandler) {
    ConcurrentKafkaListenerContainerFactory<String, Object> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory);
    factory.setCommonErrorHandler(errorHandler);
    // 添加容器异常监听器
    factory.getContainerProperties().setContainerExceptionHandler((throwable, container) -> {
        log.error("Kafka容器发生异常，容器ID：{}", container.getContainerProperties().getGroupId(), throwable);
    });
    return factory;
}

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