MongoDB Sink Connector如何指定JSON嵌套字段为_id并实现upsert

配置失效原因 #

当前配置无法满足需求的核心问题有2个：

• 主键投影路径配置错误：document.id.strategy.partial.value.projection.list=id只会读取消息根层级的id字段，无法识别嵌套在info对象下的id属性
• 策略配置冲突+缺失upsert开关：document.id.strategy错误拼接了不属于IdStrategy实现的DocumentIdAdder类，会导致主键生成逻辑异常；同时没有显式开启ReplaceOne策略的upsert开关，无法实现主键匹配即更新的逻辑

修正后的完整连接器配置 #

替换原有id策略、写入策略相关配置即可，完整可直接使用的配置如下：

topics=example
connector.class=com.mongodb.kafka.connect.MongoSinkConnector
tasks.max=1

connection.uri=mongodb://localhost:27017
database=example
collection=example

# 核心写入与主键策略配置
writemodel.strategy=com.mongodb.kafka.connect.sink.writemodel.strategy.ReplaceOneBusinessKeyStrategy
document.id.strategy=com.mongodb.kafka.connect.sink.processor.id.strategy.PartialValueStrategy
# 嵌套主键用点分隔写完整路径
document.id.strategy.partial.value.projection.list=info.id
document.id.strategy.partial.value.projection.type=AllowList
# 开启upsert：匹配到同主键文档则更新，匹配不到则插入新文档
replace.one.business.key.strategy.upsert=true
document.id.strategy.overwrite.existing=true

errors.tolerance=all
errors.deadletterqueue.topic.name=error_queue
errors.deadletterqueue.context.headers.enable=true
errors.log.include.messages=true

关键配置说明 #

• document.id.strategy.partial.value.projection.list=info.id：明确指定从消息体的info.id嵌套路径提取值，作为MongoDB文档的_id主键
• replace.one.business.key.strategy.upsert=true：开启upsert模式，是实现同主键文档更新的必要开关
• 移除原配置中错误拼接的DocumentIdAdder类，避免多策略冲突导致主键生成失败
• document.id.strategy.overwrite.existing=true：允许覆盖同_id的已有文档，避免更新时抛主键重复错误

效果验证 #

• 首次传入info.id=5的用户消息时，MongoDB会插入_id: "5"的新文档，email字段值为john@john.com
• 后续传入同info.id、email更新为john@johnmax.com的消息时，会自动匹配_id="5"的原有文档，替换后email字段同步更新为最新值，完全符合需求

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