Hi,
I have the following source and sink configurations for MongoDB Kafka connect. Create and update changes are reflected correctly in the change stream and the target database gets the modifications. However, deletions on the source mongodb database does not produce any tombstone messages and the changes are not reflected in the target DB either. Requesting assistance with this.
Source configuration:
{
"name": "del_connect_source",
"config": {
"connector.class": "com.mongodb.kafka.connect.MongoSourceConnector",
"key.converter": "org.apache.kafka.connect.storage.StringConverter",
"value.converter": "org.apache.kafka.connect.storage.StringConverter",
"connection.uri": “mongodb://localhost:27017”,
"database": "j_trials",
"collection": "kafka_connect",
"pipeline": "[{'$match': {'operationType': {'$in': ['insert', 'update', 'replace', 'delete'] }}}, {'$project': {'_id': 1,'fullDocument': 1,'ns': 1,} } ]",
"publish.full.document.only": true,
"publish.full.document.only.tombstones.on.delete": true,
"topic.prefix": "del_",
"topic.namespace.map": "{\"*\":\"j_trials.kafka_connect\"}",
"copy.existing": "true",
"tasks.max": 1,
"mongo.errors.log.enable": true
}
}
Sink configuration:
{
"name": "del_connect_sink",
"config": {
"connector.class": "com.mongodb.kafka.connect.MongoSinkConnector",
"key.converter": "org.apache.kafka.connect.storage.StringConverter",
"value.converter": "org.apache.kafka.connect.storage.StringConverter",
"topics": "del_.j_trials.kafka_connect",
"connection.uri": “mongodb://localhost:27017”,
"database": "j_trials_sink",
"collection": "kafka_connect",
"schema.enable": "false"
}
}