Issue with MongoDB sink connector, Records inserted to MongoDB are in the form of CDC structure instead of a mongo document structure

[vinaymurari]

Hi,

I am using confluent kafka connect to capture CDC from MS SQL(source) to MongoDB(sink). This is the source connecetor used for MS SQL debezium-debezium-connector-sqlserver-1.5.0. PFB source connector config.

"config":{ "connector.class": "io.debezium.connector.sqlserver.SqlServerConnector", "key.converter":"org.apache.kafka.connect.json.JsonConverter", "value.converter":"org.apache.kafka.connect.json.JsonConverter", "key.converter.schemas.enable":false, "value.converter.schemas.enable":false, "database.server.name": "test", "database.dbname": "testdb", "database.hostname": "", "database.port": "1433", "database.user": "", "database.password": "****", "database.history.kafka.bootstrap.servers": "localhost:9092", "database.history.kafka.topic": "schema_db"}

With this we are able to consume the data. However when we try to sink this data to mongoDB collection using 'hpgrahsl-kafka-connect-mongodb-1.4.0' we are able to push the data to MongoDB but in the below format.

Mongo doc structure as follows:

/**

• Paste one or more documents here */ { "op": "r", "before": null, "after": { "quantity": { "$numberLong": "150" }, "name": "banana", "id": { "$numberLong": "1" } }, "source": { "schema": "dbo", "sequence": null, "event_serial_no": null, "connector": "sqlserver", "name": "test", "commit_lsn": "00000026:000004a8:0001", "change_lsn": null, "version": "1.5.2.Final", "ts_ms": { "$numberLong": "1623065125639" }, "snapshot": "true", "db": "testdb", "table": "Inventory1" }, "ts_ms": { "$numberLong": "1623065125639" }, "transaction": null }

CDC is not handled and for each change we are getting a new record instead of updating the existing records.

Sink config as follows:

"config": { "connector.class": "at.grahsl.kafka.connect.mongodb.MongoDbSinkConnector", "topics": "test.dbo.Inventory1", "mongodb.connection.uri": "mongodb://localhost:27017/kafkaconnect?w=1&journal=true", "mongodb.collection": "mssql-moh", "mongodb.change.data.capture.handler": "at.grahsl.kafka.connect.mongodb.cdc.debezium.rdbms.RdbmsHandler", "key.converter":"org.apache.kafka.connect.json.JsonConverter", "value.converter":"org.apache.kafka.connect.json.JsonConverter", "key.converter.schemas.enable":false, "value.converter.schemas.enable":false
}

Is there a way to handle this ?

[vinaymurari]

Looks like a config issue , we will comeback with more info.

[itu-reinhard]

Looks like a config issue , we will comeback with more info.

Have you solved it? I have the same issue @vinaymurari