FlinkKafkaProducer发送Kafka事件消费后为空JSON记录问题排查

问题解答 #

1. 自定义序列化器是否需要配套自定义反序列化器

不需要。只要你序列化输出的是标准UTF-8编码的JSON字节流，Kafka控制台消费者默认会直接将字节转字符串打印，Druid的JSON摄入源也原生支持解析标准JSON结构，完全不需要额外开发反序列化器。只有当你使用非通用的自定义二进制格式（如自定义编码、Protobuf/Thrift未带Schema信息等）时，才需要配套开发对应反序列化器。

如果需要让其他系统识别自定义格式，反序列化器需要部署在对应消费侧：比如Kafka控制台消费者要使用的话需要将反序列化器打包放到Kafka的类路径下，启动命令指定反序列化器全类名；Druid要使用的话需要将反序列化器打包成Druid扩展部署到集群各节点。

2. 空JSON记录的问题根因

100%是序列化环节的问题：
你发固定字符串时消费正常，说明Flink到Kafka的链路、Kafka服务端、控制台消费者的逻辑都没问题，问题出在Jackson序列化Scala case class的步骤。
默认Jackson的ObjectMapper没有适配Scala类的特性，无法识别Scala case class的val成员变量，序列化后就会输出空的{}。解决方案是给ObjectMapper注册Scala适配模块：

// 引入Jackson Scala模块依赖后注册即可正常序列化
import com.fasterxml.jackson.module.scala.DefaultScalaModule
private val mapper = new ObjectMapper().registerModule(DefaultScalaModule)

注意ObjectMapper是线程安全的，不要在serialize方法内重复创建实例，避免性能损耗和不必要的异常。

3. Kafka 0.11.x和2.x在JSON序列化方面的差异

没有影响。Kafka服务端本身只处理字节数组，完全不感知消息体的内容格式，JSON序列化/反序列化都是客户端侧的逻辑，和Kafka服务端版本无关。只要你生产侧输出的字节格式符合消费侧的解析要求，跨版本交互没有任何问题。

4. Jackson的JSON序列化替代方案

Scala生态原生支持的JSON库普遍比Jackson适配性更好，不需要额外注册模块，常见可选方案：

• Circe：纯函数式JSON库，Scala生态使用率很高
• Play JSON：Play框架自带的JSON库，非Play项目也可单独引入
• uPickle：轻量级JSON序列化库，性能优异，配置极简
  Java生态的可选方案还有Gson、Fastjson等，对Scala类的适配性略差，适合纯Java项目使用。

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