如何通过Kafka Connect MongoDB Sink Connector完整删除MongoDB文档？

看起来你已经走对了方向，但少了关键配置项导致只能删除字段而非完整文档。让我帮你梳理下问题原因和解决步骤：

问题根源 #

你使用的DeleteOneBusinessKeyStrategy需要明确指定业务键字段，用来匹配MongoDB中要删除的文档。你的现有配置里缺少了这个关键参数，导致连接器无法正确识别要删除的目标文档，反而可能误执行了字段级操作。

解决方案步骤 #

1. 更新连接器配置，添加业务键参数 #

修改你的curl配置，在config块中加入business.key.fields参数（指定用id作为匹配字段），如果需要更明确还可以加上business.key.converter与现有转换器保持一致：

curl -X POST -H "Content-Type: application/json" -d '{
  "name":"test-testing-delete7", 
  "config":{
    "topics":"testing", 
    "connector.class":"com.mongodb.kafka.connect.MongoSinkConnector", 
    "tasks.max":"1", 
    "connection.uri":"mongodb://localhost:27017", 
    "database":"flower", 
    "collection":"testing", 
    "key.converter":"org.apache.kafka.connect.storage.StringConverter", 
    "value.converter":"org.apache.kafka.connect.storage.StringConverter", 
    "key.converter.schemas.enable":"false", 
    "value.converter.schemas.enable":"false", 
    "document.id.strategy.partial.value.projection.list":"id", 
    "document.id.strategy.partial.value.projection.type":"AllowList", 
    "business.key.fields":"id",
    "writemodel.strategy":"com.mongodb.kafka.connect.sink.writemodel.strategy.DeleteOneBusinessKeyStrategy" 
  }
}' localhost:8083/connectors

如果是更新现有连接器而非新建，用PUT请求：

curl -X PUT -H "Content-Type: application/json" -d '{
  "config":{
    # 上面完整的config内容
  }
}' localhost:8083/connectors/test-testing-delete7/config

2. 确保发送的Kafka消息格式正确 #

触发完整删除的消息只需包含匹配用的id字段即可，不要附带其他字段，避免混淆操作逻辑：

{"id":"要删除的文档ID"}

3. 可选方案：使用DeleteByIdStrategy #

如果你的MongoDB文档_id字段就是由消息中的id映射而来（你的现有document.id.strategy已经配置了这一点），也可以改用更直接的DeleteByIdStrategy，无需配置业务键，只需修改策略类：

"writemodel.strategy":"com.mongodb.kafka.connect.sink.writemodel.strategy.DeleteByIdStrategy"

这个策略会直接根据文档的_id执行删除，同样能实现完整文档删除的效果。

验证操作 #

发送测试消息到testing主题后，检查MongoDB的flower.testing集合，确认对应ID的文档已被完整移除，而不是仅删除字段。

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