1

我可以在 mongo 中插入/更新文档，但我正在努力删除文档。

这是从 SQL Server 源的 debezium 连接在 kafka 主题中记录数据的方式（最后一行是 DELETE 操作的样子）：

{"user_code":1001}      {"user_code":1001,"first_name":"Sally","last_name":"Thomas","email":"sally.thomas@acme.com"}
{"user_code":1002}      {"user_code":1002,"first_name":"George","last_name":"Bailey","email":"gbailey@foobar.com"}  
{"user_code":1003}      {"user_code":1003,"first_name":"Edward","last_name":"Walker","email":"ed@walker.com"}       
{"user_code":1003}      null

在此示例中，即使在 NULL 值之后，具有 user_code 1003 的文档仍在 MongoDB 中。

下面是我配置 MongoSinkConnector 的方式（我已经尝试了这两种方法mongodb.delete.on.null.values，delete.on.null.values但都没有奏效）：

{
  "name": "inventory-connector-sink-2",
  "config": {
      "connector.class" : "com.mongodb.kafka.connect.MongoSinkConnector",
      "tasks.max" : "1",
      "topics": "server1.dbo.customers",
      "connection.uri": "mongodb://root:root@mongo:27017/",
      "database": "testDB",
      "collection": "customers_2",
      "database.history.kafka.bootstrap.servers" : "kafka:9092",
      "database.history.kafka.topic": "schema-changes.inventory",
      "key.converter": "org.apache.kafka.connect.json.JsonConverter",
      "key.converter.schemas.enable": false,
      "value.converter": "org.apache.kafka.connect.json.JsonConverter",
      "value.converter.schemas.enable": false,
      "document.id.strategy": "com.mongodb.kafka.connect.sink.processor.id.strategy.FullKeyStrategy",
      "writemodel.strategy":"com.mongodb.kafka.connect.sink.writemodel.strategy.ReplaceOneBusinessKeyStrategy",
      "mongodb.delete.on.null.values": true
  }
}

我也尝试过使用 PartialValueStrategy 但没有运气。

PS：我正在为我的接收器连接器使用 confluentinc/cp-kafka-connect docker 映像。

4

0 回答 0