Error in ChangeStream event

Working with Data Connectors & Integrations
,
1

I am using changeStream to trigger insert and update in mongo collection.
I am using a mongo sink connector that consumes data from a kafka topic and inserts the data in the mongo collection.

On inserting this data the changeStreamEvent should get triggered, but I am getting the below error.

The ChangeStreamEvent trigger properly works when the data is inserted or updated in the mongo database using compass tool or using the mongotemplate using the java code.

Below is the kafka-mongo sink connector configuration.

ChangeStreamBug.PNG
ChangeStreamBug.PNG1820×786 464 KB

Mongo sink Connector configuration

name=mongo-sink-name
connector.class=com.mongodb.kafka.connect.MongoSinkConnector
topics=topic1, topic2
tasks.max=1
key.ignore=true
connection.uri=mongodb://uname:pass@localhost:27017/dbName
database=dbName
collection=collection1
max.num.retries=3
retries.defer.timeout=5000
type.name=kafka-connect
schemas.enable=false
document.id.strategy=com.mongodb.kafka.connect.sink.processor.id.strategy.PartialValueStratey
value.projection.list=event.serviceId,event.sender
value.projection.type=allowlist
writemodel.strategy=com.mongodb.kafka.connect.sink.writemodel.strategy.UpdateOneBusinessKeyTimestampStrategy
topic.override.topic2.collection=collection2
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false
key.converter=org.apache.kafka.connect.storage.StringConverter
key.converter.schemas.enable=false
transforms=WrapKey
transforms.WrapKey.type=org.apache.kafka.connect.transforms.HoistField$Key
transforms.WrapKey.field=_id
change.data.capture.handler=com.mongodb.kafka.connect.sink.cdc.mongodb.ChangeStreamHandler
2

I thjink the CDC handler expects the kafka topic message to resemble a change stream event, since you are changing the message via Hoist event and project the message is no longer in the format. To test this simplify the CDC sink to something like

 "tasks.max":"1",
 "topics":"topic1",
"change.data.capture.handler":"com.mongodb.kafka.connect.sink.cdc.mongodb.ChangeStreamHandler",
  "connection.uri":"mongodb://uname:pass@localhost:27017/dbName",
  "database":"dbName",
"collection":"collection1"```