使用Debezium实现Postgres间数据同步失败问题求助

一、检查Debezium源连接器状态与日志 #

• 用Kafka Connect API确认连接器状态：

  curl -X GET http://<connect-host>:<connect-port>/connectors/<source-connector-name>/status
  

  确保状态为RUNNING，若为FAILED或PAUSED，重点分析日志错误
• 查看源连接器日志，重点排查：
  • 是否有publication相关错误，比如无法访问dbz_publication
  • 是否成功创建/识别逻辑复制槽（日志含Created replication slot或Found existing replication slot）
  • 是否正确监控public.mytable（日志含Adding table to monitored set: public.mytable）
  • 权限类错误（如permission denied for replication slot、cannot select from table mytable）

二、验证Postgres源端配置 #

• 检查postgresql.conf核心参数：

  wal_level = logical       # 必须设置为logical，不能是replica/minimal
  max_replication_slots = 1 # 至少留1个槽位给Debezium
  max_wal_senders = 1       # 至少1个WAL发送进程
  

• 确认dbz_publication包含目标表：

  SELECT * FROM pg_publication_tables WHERE pubname = 'dbz_publication';
  

  确保结果中存在public.mytable
• 检查源库用户权限：

  -- 确认用户有REPLICATION权限
  SELECT rolname, rolreplication FROM pg_roles WHERE rolname = '<connector-user>';
  -- 确认用户对mytable有SELECT权限
  SELECT has_table_privilege('<connector-user>', 'public.mytable', 'SELECT');
  

三、检查Kafka主题状态 #

• 确认Debezium生成的主题存在（默认格式：{server-name}.{schema}.{table}）：

  kafka-topics.sh --list --bootstrap-server <kafka-host>:<kafka-port>
  

• 如果主题存在，尝试消费数据验证：

  kafka-console-consumer.sh --bootstrap-server <kafka-host>:<kafka-port> \
    --topic <target-topic-name> --from-beginning
  

• 若主题不存在，说明源连接器未正常初始化，回到第一步排查

四、排查JDBC Sink连接器配置与状态 #

• 确认Sink连接器状态：

  curl -X GET http://<connect-host>:<connect-port>/connectors/<sink-connector-name>/status
  

• 检查关键配置项：
  • auto.create=true：确保开启自动创建目标表
  • connection.url：确认目标Postgres的连接地址、库名正确
  • topics：必须和源连接器生成的主题名完全一致，无拼写错误
  • key.converter/value.converter：需匹配源连接器的格式（比如源用Debezium转换器，Sink用org.apache.kafka.connect.json.JsonConverter并设置schemas.enable=false）
• 查看Sink日志，重点找：
  • 目标库连接失败错误（如Connection refused、invalid credentials）
  • 主题不存在错误（说明topics配置错误）
  • 权限不足（目标库用户无CREATE TABLE权限）
  • 数据格式解析错误（转换器不兼容）

五、手动触发数据测试 #

• 在源库插入测试数据：

  INSERT INTO mytable(id, message) VALUES(1, 'test_sync');
  

• 再次检查Kafka主题是否有新数据，同时查看源、Sink连接器日志是否有对应操作记录

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