如何避免Kafka Connect JDBC源连接器增量模式下丢记录？

你遇到的核心问题是增量ID同步逻辑与事务提交顺序不匹配：当小ID事务晚于大ID事务提交时，连接器会因为已记录的最大ID大于小ID而永久跳过该记录。以下是无需切换到全CDC方案的可靠解决思路：

方案1：自定义查询逻辑，引入回溯检查窗口 #

通过自定义查询替代连接器自动生成的SQL，在每次增量同步时，额外回溯检查上一个周期内可能遗漏的延迟提交记录。

配置步骤： #

1. 在数据库中创建一个同步状态表，用于存储关键同步状态：

   CREATE TABLE sync_state (
       connector_name VARCHAR(100) PRIMARY KEY,
       last_synced_max_id BIGINT NOT NULL,
       last_backtrack_id BIGINT NOT NULL DEFAULT 0
   );
   

2. 配置JDBC源连接器使用自定义query参数，SQL逻辑示例：

   -- 1. 回溯检查上一窗口内延迟提交的记录
   SELECT * FROM your_table 
   WHERE id > (SELECT last_backtrack_id FROM sync_state WHERE connector_name = 'your-jdbc-connector')
     AND id <= (SELECT last_synced_max_id FROM sync_state WHERE connector_name = 'your-jdbc-connector')
   UNION
   -- 2. 增量同步新生成的记录
   SELECT * FROM your_table 
   WHERE id > (SELECT last_synced_max_id FROM sync_state WHERE connector_name = 'your-jdbc-connector');
   

3. 配合定时任务（或连接器扩展），定期将last_backtrack_id更新为当前的last_synced_max_id（例如每10次轮询更新一次），避免回溯窗口无限扩大。

优缺点： #

• 优点：无需修改业务逻辑，能覆盖延迟提交的事务；
• 缺点：需要维护状态表，自定义SQL复杂度较高，回溯会增加数据库查询负载。

方案2：改用timestamp+incrementing复合模式，基于事务提交时间戳 #

如果能在表中添加事务提交时间戳列（而非应用设置的创建时间），可以结合时间戳和ID的复合逻辑，确保延迟提交的记录能被后续轮询捕获。

配置步骤： #

1. 给业务表添加commit_time列，通过应用逻辑确保该列记录事务提交时间（MySQL无内置提交时间触发器，需在事务提交前由应用更新该列）；
2. 配置连接器核心参数：

   mode=timestamp+incrementing
   timestamp.column.name=commit_time
   incrementing.column.name=ID
   table.poll.interval.ms=30000
   validate.non.null=true
   

核心逻辑： #

连接器会使用commit_time > last_timestamp OR (commit_time = last_timestamp AND ID > last_id)作为查询条件，只要延迟提交的事务commit_time晚于上一次轮询的时间戳，就会被捕获。

优缺点： #

• 优点：利用连接器原生功能，无需复杂自定义；
• 缺点：需要改造表结构和写入逻辑，确保commit_time的准确性。

方案3：强制事务提交顺序与ID生成顺序一致 #

从业务写入端解决问题，确保小ID的事务先于大ID的事务提交，从根源避免顺序不一致的情况。

实现方式： #

• 应用层引入分布式锁，确保按ID递增顺序提交插入事务；
• 批量插入场景拆分事务为单条插入并按ID顺序提交（仅适用于低并发场景）。

优缺点： #

• 优点：彻底解决问题，无需修改连接器配置；
• 缺点：对应用侵入性强，可能影响写入性能。

方案4：增量同步+定期全量补数 #

结合增量同步和定期全量同步，用全量同步弥补增量同步的丢数问题，要求下游系统支持幂等处理。

配置步骤： #

1. 主连接器使用incrementing模式进行常规增量同步；
2. 部署第二个JDBC源连接器，使用bulk模式，配置较长的table.poll.interval.ms（例如86400000，即每天一次），全量同步整个表；
3. 下游消费者通过ID去重（如将ID作为Kafka消息的key，利用Kafka幂等性或下游存储的唯一约束）。

优缺点： #

• 优点：配置简单，无需复杂改造；
• 缺点：全量同步会增加数据库和Kafka负载，依赖下游幂等能力。

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