Kafka反序列化异常时如何获取并记录原始消息？同时解决handleOne方法未被触发的问题

嗨，我来帮你搞定这两个问题！咱们一步步拆解解决：

一、先解决handleOne方法未被触发的问题 #

你当前遇到的handleOne未执行、反而走handleOtherException的情况，核心原因是：直接使用JsonDeserializer时，反序列化异常发生在Kafka Consumer拉取消息的前置阶段，还没到调用监听器的环节，容器会把这类异常归类为“其他异常”处理。

要让反序列化异常能被handleOne捕获，你需要用ErrorHandlingDeserializer包装你的JsonDeserializer——它会把反序列化异常封装成带原始ConsumerRecord的ListenerExecutionFailedException，这样就能触发handleOne方法了。

修改你的消费者配置如下：

@Bean
public ConsumerFactory<String, EventEnvelop> consumerConfigs() {
    Map<String, Object> configs = new HashMap<>();
    configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddresses);
    configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    // 用ErrorHandlingDeserializer包装JsonDeserializer
    configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
    // 指定实际的Json反序列化实现类
    configs.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, JsonDeserializer.class.getName());
    configs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

    log.debug("kafka consumer bootstrap addresses: {}", bootstrapAddresses);
    configs.forEach((key, value) -> log.debug("kafka consumer configuration: {\"{}\": \"{}\"", key, value));

    return new DefaultKafkaConsumerFactory<>(configs);
}

调整后，反序列化异常就会被正确封装，流转到handleOne方法中处理了。

二、获取并记录原始的未解析消息 #

当你用上ErrorHandlingDeserializer后，就能从异常对象里拿到原始的消息字节数据了。RecordDeserializationException提供了getData()方法，返回的就是导致反序列化失败的原始字节数组，你可以把它转成字符串记录下来。

修改你的KafkaConsumerErrorHandler的handle方法：

private boolean handle(Exception exception, Consumer<?,?> consumer) {
    // 先判断是否是监听器执行异常，内部包裹了反序列化异常
    if (exception instanceof ListenerExecutionFailedException lefe) {
        Throwable rootCause = lefe.getCause();
        if (rootCause instanceof RecordDeserializationException e) {
            // 将原始字节转成字符串（编码可根据你的实际消息格式调整，这里用UTF-8）
            String rawMessage = new String(e.getData(), StandardCharsets.UTF_8);
            log.error("无法解析的原始消息内容: {}", rawMessage);
            log.error("解析失败原因: {}", e.getMessage());
            
            // 跳过这条坏消息并提交偏移量
            consumer.seek(e.topicPartition(), e.offset() + 1L);
            consumer.commitSync();
        }
    } else if (exception instanceof RecordDeserializationException e) {
        // 兼容直接捕获到反序列化异常的情况
        String rawMessage = new String(e.getData(), StandardCharsets.UTF_8);
        log.error("无法解析的原始消息内容: {}", rawMessage);
        log.error("解析失败原因: {}", e.getMessage());
        
        consumer.seek(e.topicPartition(), e.offset() + 1L);
        consumer.commitSync();
    } else {
        log.error("处理消息时发生意外错误: ", exception);
    }
    return false;
}

这样就能成功记录下那条导致反序列化失败的原始消息了！

额外说明（关于消息顺序） #

你通过setConcurrency(1)和setBatchListener(false)来保证消息顺序的做法是完全正确的——单线程消费同一个分区的消息，能严格保证消息的消费顺序，这个配置不需要调整。

备注：内容来源于stack exchange，提问作者zappee