Kafka Connect JDBC Sink读取Kafka Topic数据异常问题求助

核心排查方向 #

1. 确认Debezium CDC消息的解析配置 #

Debezium采集的MySQL数据是嵌套结构（包含before/after/source等层级），JDBC Sink默认无法直接识别after里的业务数据，这是最常见的问题根源：

• 检查JDBC Sink Connector是否配置了unwrap转换，必须添加以下配置项：

  transforms=unwrap
  transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState
  transforms.unwrap.drop.tombstones=false
  transforms.unwrap.delete.handling.mode=rewrite
  

  这个转换会把CDC消息里的after字段提取出来作为顶层数据，让JDBC Sink能识别到email、password等业务字段。

2. 检查JDBC Sink的表结构适配配置 #

• 若启用auto.create自动建表，需同时开启auto.evolve=true，否则Connector只会根据初始识别到的字段（通常只有主键）创建表，后续不会自动添加新字段。
• 确认pk.mode=record_key或pk.mode=record_value，并正确设置pk.fields=id，确保主键映射正确，避免字段识别异常。

3. 验证目标MySQL表的约束与消息内容匹配 #

• 预建表时报Field 'email' doesn't have a default value，本质是JDBC Sink没有向该字段写入值，而表字段设置为NOT NULL且无默认值。结合自动建表只生成id的现象，说明Sink根本没读取到email、password字段的数据，核心还是消息结构未被正确解析。
• 可以临时给测试表的非主键字段添加默认值（如email VARCHAR(255) DEFAULT '' NOT NULL），再运行Sink验证：如果能写入空字符串，就坐实了消息解析的问题。

4. 直接查看Topic中的消息内容 #

用Kafka命令行工具消费Topic，确认消息结构是否正确：

kafka-console-consumer.sh --bootstrap-server <你的Kafka地址> --topic <目标Topic名> --from-beginning --property print.key=true --property value.deserializer=org.apache.kafka.common.serialization.StringDeserializer

如果输出的消息里after字段包含完整的id、email、password，那就是Sink的解析配置有问题；如果after字段本身缺失数据，就要回头检查Debezium CDC的采集配置。

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