通过Debezium JDBC Sink Connector连接Kafka与PostgreSQL异常排查

1. 确认Kafka Connect集群与Connector运行状态 #

• 检查Kafka Connect Pod是否正常运行：

  oc get pods -n amq-streams | grep postgres-connect
  

• 查看Connector运行状态，确认任务是否启动：

  oc get kafkaconnectors jdbc-postgres-sink-connector -n amq-streams -o yaml
  

  重点看status字段里的connectorStatus和tasks状态，若显示FAILED，直接提取错误信息定位问题。

2. 验证Kafka主题消息投递情况 #

• 确认发送的消息确实进入postgres主题：

  # 用kcat或kafka-console-consumer消费主题消息
  kcat -b <AMQ Streams Kafka URL> -t postgres -C -e -o beginning
  

  注意客户端要配置正确的TLS证书才能连接集群。如果看不到发送的消息，问题出在生产环节，和Connect无关。

3. 核对Protobuf Schema与数据库字段匹配性 #

• 检查Schema Registry中的Protobuf schema和PostgreSQL Authors表的字段：
  • 字段大小写：PostgreSQL默认大小写不敏感，但Connector可能严格匹配，比如配置里primary.key.fields: id，但数据库表主键是Id，大小写不匹配会导致主键识别失败。
  • 字段类型：确保Protobuf字段类型和数据库兼容，比如int32对应integer、string对应varchar等。

4. 检查JDBC Sink Connector关键配置 #

• 主键配置：当前primary.key.mode: record_key意味着从消息Key中取id作为主键，要确认发送的消息Key里确实包含id字段且类型正确。如果主键存在于Value中，需改为primary.key.mode: record_value并调整对应字段配置。
• 连接权限：验证connection.url的地址、端口、库名是否正确，connection.user是否拥有Authors表的INSERT/UPDATE权限。可以在Connect Pod内测试连接：

  oc exec -it <postgres-connect-pod-name> -n amq-streams -- bash
  # 用psql测试JDBC连接
  psql "jdbc:postgresql://<Postgres URL>:5432/sampledb" -U postgres
  

• 插入模式：insert.mode: upsert依赖主键才能正常工作，若消息无主键或主键与数据库不匹配，会导致插入失败。

5. 查看Kafka Connect日志 #

• 拉取Connect Pod日志，定位错误信息：

  oc logs <postgres-connect-pod-name> -n amq-streams -f
  

  重点关注含jdbc、sink、error的日志条目，常见问题包括Schema转换错误、JDBC连接失败、字段类型不兼容、主键缺失等。

6. 检查插件加载情况 #

• 确认Debezium JDBC Connector和Protobuf Converter插件已正确加载：

  oc exec -it <postgres-connect-pod-name> -n amq-streams -- ls /opt/kafka/plugins/
  

  应能看到kafka-connect-protobuf-converter和debezium-jdbc-connector相关jar包，若缺失，检查Kafka Connect的build配置中插件下载地址是否正确，镜像是否重新构建推送。

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