These days, I was doing the experiments to test the capability of Kafka connector from Mongo to Kafka queue.

The issue was triggered when I use the unwrap transfrom.

name=...
connector.ckass=io.debezium.connector.mongodb.MongoDbConnector

mongodb.hosts=...
mongodb.name=...

transfroms=unwrap
transforms.unwarp.type=io.debezium.connector.mongodb.transforms.UnwrapFromMongoDbEnvelope.

When I run the connector, the console throws an unrecoverable BSON exception.
For testing purpose, I comment out the last two lines.
I got the following json data in Kafka Queue

{"id":"{ "$oid" : "5a861110091252703ce2eb29"}"} {"after":null,"patch":{"string":"{"$v" : 1,"$set" : {"name" : "test"}}"},"source":{"name":"kconnect_db","rs":"rs0","ns":"test.kconnect_test","sec":1518735770,"ord":1,"h":{"long":5293348459877549762},"initsync":null},"op":{"string":"u"},"ts_ms":{"long":1518735770420}}

But in line 53 of io.debezium.connector.mongodb.transforms.UnwrapFromMongoDbEnvelope.java

"valueDocument = valueDocument.getDocument("$set");"

It directly get the value of "$set" from "patch" section. Obviously it was incorrect, it should firstly get the document from "string", then get the value from the "$set"