JDBC Sink Connector无法消费PostgreSQL Delete操作问题求助

1. 检查Debezium源连接器核心配置 #

• 确认PostgreSQL的wal_level设置为logical，这是CDC同步的基础条件。在源数据库执行以下SQL验证：

  SHOW wal_level;
  

  若结果不是logical，需修改postgresql.conf并重启数据库。
• 显式指定Debezium解码插件为pgoutput（现代推荐插件，避免旧插件的兼容性问题），在源连接器配置中添加：

  "plugin.name": "pgoutput"
  

• 验证源端是否生成墓碑消息：用Kafka命令行工具消费目标topic，检查是否存在value为null的记录（墓碑消息）：

  kafka-console-consumer.sh --bootstrap-server <kafka-host>:9092 --topic utanga_dev.public.exchange_sellerportcharge --from-beginning --property print.key=true --property print.value=true
  

  若未发现墓碑消息，需排查Debezium日志及数据库权限（确保数据库用户拥有REPLICATION权限和目标表的SELECT权限）。

2. 修复JDBC Sink连接器的转换配置 #

你的Sink配置中使用了ExtractNewRecordState转换，但默认该转换会丢弃墓碑消息，导致Sink无法接收删除事件。需添加转换参数保留并处理墓碑消息：

"transforms.unwrap.delete.handling.mode": "rewrite"

修改后的transforms配置段应为：

"transforms": "unwrap",
"transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
"transforms.unwrap.delete.handling.mode": "rewrite"

该配置会将墓碑消息转换为仅包含主键字段、其余字段为null的记录，让JDBC Sink能够识别并执行删除操作。

3. 验证Sink连接器主键配置 #

• 确认pk.fields指定的id是目标表的真实主键，且Kafka消息的key中正确包含该字段（可通过消费topic查看key内容验证）。
• 确保insert.mode保持为upsert（你已配置），delete.enabled仅在upsert模式下生效。

4. 检查连接器日志 #

• 查看Debezium源连接器日志，确认删除事件被捕获并发送到Kafka：

  docker logs <debezium-connector-container> | grep "delete"
  

• 查看JDBC Sink连接器日志，确认是否接收到墓碑消息或删除相关记录：

  docker logs <jdbc-sink-connector-container> | grep "delete"
  

  若日志中无删除相关内容，需排查Kafka topic权限、网络连通性等问题。

5. 其他注意事项 #

• 确保Debezium与JDBC Sink版本兼容：优先使用Confluent Platform或Debezium的稳定版本组合，避免版本不匹配引发的问题。
• 检查目标数据库权限：Sink使用的数据库用户需拥有目标表的DELETE权限。

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