Debezium与Kafka MongoDB Sink Connector时序文档更新新增记录问题

针对你遇到的问题——配置timeseries.timefield后MongoDB Sink Connector强制使用插入策略，导致PostgreSQL更新操作在MongoDB中新增文档而非更新原有记录，有以下几种可行的解决思路：

1. 利用MongoDB唯一索引+Upsert配置 #

这是最简便的无代码方案：

• 第一步：在MongoDB时序集合上创建基于PostgreSQL主键（或唯一业务键）的唯一索引。假设你的PostgreSQL表主键为id，执行MongoDB命令：

  db.your_timeseries_collection.createIndex({ id: 1 }, { unique: true })
  

  若业务键需与时序时间字段组合才能保证唯一，可创建复合唯一索引：

  db.your_timeseries_collection.createIndex({ id: 1, timestamp: 1 }, { unique: true })
  

• 第二步：在MongoDB Sink Connector配置中添加：

  mongodb.upsert=true
  mongodb.document.id=id  # 指定PostgreSQL主键字段作为MongoDB文档的_id，或依赖上述创建的唯一索引
  

  配置后，连接器执行插入操作时若遇到唯一键冲突，会自动转为更新操作，覆盖原有文档。

2. 自定义写入策略（需扩展连接器） #

若需要更精细的更新逻辑，可自定义MongoDB Sink Connector的WriteStrategy：

• 继承官方的InsertOneDefaultStrategy，重写process方法：
  • 判断事件类型：若为Debezium捕获的UPDATE事件，调用MongoDB的updateOne方法，通过唯一键匹配文档并更新
  • 若为INSERT事件，保持原有的insertOne逻辑
• 编译自定义策略类，打包后放入Kafka Connect的插件目录，再在连接器配置中指定：

  mongodb.write.strategy=com.your.package.CustomTimeseriesWriteStrategy
  

3. 绕过官方时序自动配置，手动处理时序写入 #

放弃使用连接器的timeseries.timefield配置，自行保证写入合规性：

• 确保Debezium捕获的事件包含MongoDB时序集合要求的timeField（如timestamp）
• 不配置连接器的timeseries.*参数，使用默认更新策略（此时连接器会对Debezium的UPDATE事件执行updateOne操作）
• 提前在MongoDB中创建好时序集合，确保所有写入文档都包含指定的timeField且符合时序约束

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