Kafka使用JsonDeserializer遇非JSON消息致消费中断该如何处理

你遇到的崩溃问题根源是org.apache.kafka.connect.json.JsonDeserializer在遇到无法解析的非JSON格式数据时，会直接抛出未捕获的异常终止消费流程，Kafka默认没有处理反序列化阶段的异常，会直接退出。可以通过以下几种方案解决：

方案1：使用官方提供的错误处理反序列化器包装（无需改业务代码，推荐） #

Kafka官方自带ErrorHandlingDeserializer，可以包装实际的反序列化器，自动捕获反序列化异常并返回null，避免进程崩溃。
配置示例如下：

# 原生Kafka消费者配置
value.deserializer=org.apache.kafka.common.serialization.ErrorHandlingDeserializer
value.deserializer.delegate.class=org.apache.kafka.connect.json.JsonDeserializer

# 如果是Spring Kafka环境，配置为
spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
spring.kafka.consumer.properties.spring.deserializer.value.delegate.class=org.apache.kafka.connect.json.JsonDeserializer

配置完成后，反序列化失败的消息会返回null值，你只需要在业务消费逻辑里判断如果value为null，就跳过该条消息，或者做归档处理，继续消费后续消息即可。

方案2：自定义安全反序列化器（灵活度最高） #

你可以继承原生JsonDeserializer，手动捕获所有反序列化异常，自定义异常处理逻辑：

import org.apache.kafka.connect.json.JsonDeserializer;

public class SafeJsonDeserializer<T> extends JsonDeserializer<T> {
    @Override
    public T deserialize(String topic, byte[] data) {
        // 兼容空消息的特殊判断
        if (data == null || data.length == 0) {
            // 可以返回默认值，或者直接返回null
            return null;
        }
        try {
            return super.deserialize(topic, data);
        } catch (Exception e) {
            // 此处可以添加日志，记录非法消息的topic、内容等信息用于排查
            return null;
        }
    }
}

之后将消费者的value.deserializer配置为你自定义的SafeJsonDeserializer全类名即可。

方案3：全局异常捕获（适合Spring Kafka用户） #

如果使用Spring Kafka框架，可以直接配置全局CommonErrorHandler，统一处理包括反序列化异常在内的所有消费异常，可配置策略为跳过非法消息、重试、转存死信队列等：

@Bean
public CommonErrorHandler kafkaErrorHandler() {
    DefaultErrorHandler errorHandler = new DefaultErrorHandler(
            // 遇到异常直接跳过当前消息，不重试
            new FixedBackOff(0, 0)
    );
    // 可以添加逻辑，将异常消息转存到死信队列
    return errorHandler;
}

补充建议 #

• 所有处理非法消息的逻辑都建议留存日志，记录对应消息的topic、分区、偏移量，方便回溯上游异常问题。
• 条件允许的情况下，可以搭建死信队列，将所有处理失败的消息统一归档存储，避免数据意外丢失。

内容的提问来源于stack exchange，提问作者SRI HARSHA S V S