PostgreSQL迁移后如何重置Debezium偏移量以从新服务器流式传输

针对你提到的PostgreSQL并行升级后，Debezium连接器对接新库时因WAL位置不匹配无法继续同步的问题，有两种可靠的方式来重置偏移量/指定新的WAL位置，无需重新执行快照：

方法一：手动修改Kafka Connect的偏移量存储 #

1. 停止目标Debezium连接器，避免运行中偏移量被覆盖
2. 查询连接器当前偏移量：使用Kafka命令行工具找到connect-offsets主题中对应连接器的记录，命令示例：

   kafka-console-consumer.sh --bootstrap-server <Kafka集群地址> --topic connect-offsets --from-beginning --property print.key=true --property print.value=true | grep <你的连接器名称>
   

3. 修改或删除偏移量记录：
   • 先在新PostgreSQL实例执行SELECT pg_current_wal_lsn();获取当前LSN
   • 编辑偏移量记录中的lsn字段为新的LSN值，或者直接删除该条记录（重启连接器后会自动以新库当前LSN开始消费）
4. 更新连接器配置：将database.hostname、database.port等参数修改为新服务器信息，添加snapshot.mode=never强制跳过快照
5. 重启连接器，验证新库的变更是否正常流入Kafka

方法二：使用Debezium信号（Signal）机制 #

适用于Debezium 1.9及以上版本，无需直接操作偏移量主题：

1. 确认信号主题配置：确保连接器配置中指定了signal.enabled=true（默认开启），信号主题默认是debezium-signals
2. 发送重置偏移量信号：向信号主题发送以下JSON格式消息，指定新库的当前LSN：

   {
     "type": "reset-offset",
     "data": {
       "connectorName": "<你的连接器名称>",
       "offset": {
         "server": "<连接器配置的server.name值>",
         "lsn": "<新库当前LSN>"
       }
     }
   }
   

3. 更新连接器配置：修改连接参数指向新服务器，添加snapshot.mode=never
4. 重启连接器，它会读取信号指令，从指定的LSN开始同步新库的变更

关键注意事项 #

• 操作前必须确保新旧PostgreSQL实例的逻辑复制已经完成同步，旧库所有未同步的变更都已推送到新库，避免数据断层
• 新库需提前配置好逻辑复制槽，Debezium连接器账号需拥有REPLICATION权限及目标表的读写权限
• 完成配置后，需检查Kafka主题中的消息是否连续，验证同步状态正常

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