KafkaListener的errorHandler未捕获异常，求配置解决方案

在@KafkaListener中配置了自定义ConsumerAwareListenerErrorHandler，但保存MongoDB时触发的DuplicateKeyException未被该处理器捕获，而是由InvocableHandlerMethod.doInvoke()捕获，导致输出冗长堆栈信息。需要排查配置问题并实现正确的异常捕获与日志记录。

现有代码片段 #

Kafka 消费方法 #

@KafkaListener(topics="myTopic", groupId="myGroup", errorHandler="myErrorHandler")
public void consume(ConsumerRecord<String, MyRecord> consumerRecord) {
  myDbService.saveRecord(consumerRecord.value());
}

自定义错误处理器 #

public class MyErrorHandler implements ConsumerAwareListenerErrorHandler {
  @Override
  public Object handleError(....) {
    log.error("Error!!!");
  }
}

异常信息 #

异常链 #

MongoWriteException > DuplicateKeyException > ListenerExecutionFailedException > KafkaException

异常片段 #

Error handler threw an exception
org.springframework.kafka.KafkaException: Seek to current after exception; nested exception is org.springframework.kafka.listener.ListenerExecutionFailedException: Listener method 'public void com.myapp.myconsumer.consume(org.apache.kafka.clients.consumer.ConsumerRecord<java.lang.String, com.myapp.domain.MyRecord>)' threw exception; nested exception is org.springframework.dao.DuplicateKeyException: Write operation error on server somehost.com:29000. Write error: WriteError{code=11000, message='.....'', details={}}.; nested exception is com.mongodb.MongoWriteException: Write operation error on server .....

1. 确保错误处理器被Spring容器托管 #

自定义MyErrorHandler必须被Spring识别为Bean，否则@KafkaListener中指定的errorHandler="myErrorHandler"无法找到对应实例。可通过两种方式注册：

• 在处理器类上添加@Component注解（指定Bean名称与errorHandler属性一致）：

@Component("myErrorHandler")
public class MyErrorHandler implements ConsumerAwareListenerErrorHandler {
    // 实现代码
}

• 或在配置类中手动声明Bean：

@Configuration
public class KafkaConfig {
    @Bean("myErrorHandler")
    public ConsumerAwareListenerErrorHandler myErrorHandler() {
        return new MyErrorHandler();
    }
}

2. 完善handleError方法的实现 #

当前handleError参数不全且未处理嵌套异常，需完整实现逻辑以提取并处理根异常：

import org.springframework.kafka.listener.ConsumerAwareListenerErrorHandler;
import org.springframework.kafka.listener.ListenerExecutionFailedException;
import org.springframework.dao.DuplicateKeyException;
import org.apache.kafka.clients.consumer.Consumer;
import org.springframework.messaging.Message;
import org.springframework.kafka.support.KafkaHeaders;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@Component("myErrorHandler")
public class MyErrorHandler implements ConsumerAwareListenerErrorHandler {
    private static final Logger log = LoggerFactory.getLogger(MyErrorHandler.class);

    @Override
    public Object handleError(Message<?> message, ListenerExecutionFailedException exception, Consumer<?, ?> consumer) {
        // 提取根异常
        Throwable rootCause = exception.getRootCause();
        if (rootCause instanceof DuplicateKeyException) {
            // 针对重复键异常打印简洁日志
            String topic = (String) message.getHeaders().get(KafkaHeaders.RECEIVED_TOPIC);
            Long offset = (Long) message.getHeaders().get(KafkaHeaders.OFFSET);
            log.error("MongoDB重复键异常: 记录主键重复，topic: {}, offset: {}", topic, offset);
        } else {
            // 其他异常按需处理，可选择打印完整堆栈或自定义日志
            log.error("消费消息时发生未知错误", exception);
        }
        // 返回null表示不触发重试（默认会将offset重置到当前位置，可根据业务调整）
        return null;
    }
}

3. 检查容器异常处理配置（可选） #

若异常仍未被捕获，需确认Kafka监听容器的错误处理器配置，避免默认处理器拦截异常：

@Configuration
public class KafkaConfig {
    @Bean
    public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
            ConsumerFactory<Object, Object> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<Object, Object> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        // 显式设置自定义错误处理器
        factory.setErrorHandler(myErrorHandler());
        return factory;
    }
}

4. 验证异常捕获逻辑 #

确保handleError方法正确接收ListenerExecutionFailedException，并通过getRootCause()获取到底层的DuplicateKeyException，这样就能针对性打印简洁日志，避免输出冗长堆栈。

内容的提问来源于stack exchange，提问作者αƞjiβ