使用 MongoDB Sink Connector 通过不同的主键更新现有文档

Question

我正在尝试通过 Confluent Cloud 设置一个 MongoDB 接收器连接器，它可以在 pgsql 和 MongoDB 之间保持数据同步。

我希望下面的配置基于 id (int) 字段（不是 _id - objectId）更新现有文档，但是它只是在 MongoDB 中创建了一个新文档消费。来自 pg 的文档将不包含 _id 字段，因此我们需要在我们的 pgsql 主键 (id) 上完成查找。

知道为什么这不像我预期的那样有效吗？

 {
  "connector.class": "MongoDbAtlasSink",
  "name": "mongodb-sink",
  "kafka.api.key": "mykey",
  "kafka.api.secret": "mysecret",
  "input.data.format": "JSON",
  "topics":"mytopic",
  "connection.host": "myhost",
  "connection.user": "myuser",
  "connection.password": "mypassword",
  "database": "mydatabase",
  "delete.on.null.values": "false",
  "tasks.max": "1",
  "document.id.strategy": "com.mongodb.kafka.connect.sink.processor.id.strategy.PartialValueStrategy",
  "document.id.strategy.partial.value.projection.list":"id",
  "document.id.strategy.partial.value.projection.type":"AllowList",
"writemodel.strategy":"com.mongodb.kafka.connect.sink.writemodel.strategy.ReplaceOneBusinessKeyStrategy"
}

Answer 1

key.projection.type 和 value.projection.type 显然在云版本中不受支持，因此它不适用于该配置。

Cloud Connector Limitations

使用 MongoDB Sink Connector 通过不同的主键更新现有文档

Using MongoDB Sink Connector to update existing documents by a different primary key

mongodb

apache-kafka

confluent-cloud

confluent-platform