Kafka Connect JDBC报JsonConverter异常求助：value为null问题排查

咱们先梳理下你遇到的核心问题：Kafka Connect的JsonConverter抛出DataException，提示需要schema和payload字段，同时调试发现toConnectData方法中的value参数为null。结合你的数据流架构和配置，咱们从几个关键方向逐一排查：

1. 补全Kafka Connect Sink的转换器配置 #

你的Sink连接器配置里缺少了key和value的转换器设置，这是最容易忽略的核心问题！默认情况下Kafka Connect可能使用的不是JsonConverter，或者即使使用了，也没有开启匹配你消息格式的schema模式。

你需要在config节点中添加以下配置项：

"key.converter": "org.apache.kafka.connect.json.JsonConverter",
"key.converter.schemas.enable": "false",
"value.converter": "org.apache.kafka.connect.json.JsonConverter",
"value.converter.schemas.enable": "true"

简单解释下：

• 你的消息value是标准的{"schema": ..., "payload": ...}结构，所以value.converter.schemas.enable设为true，和消息格式完全匹配
• 通常消息key没有复杂schema结构，所以设key.converter.schemas.enable=false即可；如果你的key也是带schema的结构，对应调整这个值就行

补全后的完整Sink配置如下：

{
  "name": "user-sink",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "tasks.max": "1",
    "topics": "user",
    "connection.url": "jdbc:mysql://localhost:3306/my_db?verifyServerCertificate=false",
    "connection.user": "root",
    "connection.password": "root",
    "auto.create": "true",
    "insert.mode": "upsert",
    "pk.fields": "id",
    "pk.mode": "record_value",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "key.converter.schemas.enable": "false",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "true"
  }
}

2. 排查Flink写入Kafka的消息是否存在空值 #

调试中发现value参数为null，说明Kafka Connect接收到了空消息体的记录。你需要做两步验证：

• 用Kafka控制台消费者查看user主题的所有消息，确认是否存在空消息体，同时检查正常消息的结构是否和你提供的示例一致：

  kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic user --from-beginning
  

• 回到Flink代码中排查：是否在处理逻辑中生成了空的Row对象，或者序列化环节出现异常，导致写入Kafka的消息体为null。比如如果Flink的Kafka生产者没有正确配置序列化器，就可能出现这种情况。

3. 确认Flink序列化输出的消息格式一致性 #

你的数据流中Debezium输出的是带Schema的消息，Flink消费后写入Kafka的格式必须和Kafka Connect的转换器要求匹配：

• 如果你用Flink的JsonSchemaSerializationSchema序列化消息，要确保它生成的是标准的{"schema": ..., "payload": ...}结构，和你提供的示例完全一致
• 避免在Flink处理中误修改消息结构，比如不小心把payload直接作为消息体发送（这种情况就需要把schemas.enable设为false，但你的示例是带schema的，所以必须保持结构一致）

4. 检查Kafka Connect的全局转换器配置 #

如果你的Sink连接器没有单独指定转换器，Kafka Connect会使用全局的key.converter和value.converter配置。你可以检查Connect的worker配置文件（比如connect-distributed.properties），确认全局转换器的schemas.enable设置是否和你的消息格式匹配。如果全局配置是schemas.enable=false，而你的消息是带schema的结构，就会触发这个异常。

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