Kafka connector stuck with MongoDB Change Stream error

Working with Data Connectors & Integrations
,
1

Hello Team,

We encountered following error when were doing Kafka MongoDB Source Connector:

[2021-03-02 07:15:33,282] INFO An exception occurred when trying to get the next item from the Change Stream: Command failed with error 10334 (BSONObjectTooLarge): ‘BSONObj size: 20112313 (0x132E3B9) is invalid. Size must be between 0 and 16793600(16MB) First element: _id: { _data: “82603CB5170000001B2B022C0100296E5A10042F63D0C3B388496786DFA01F1DD6E98B46645F69640064603C9CFEFA49CA266000FDBE0004” }’ on server xxx3.xxx.xxx.internal:27017. The full response is {“operationTime”: {"$timestamp": {“t”: 1614669333, “i”: 23}}, “ok”: 0.0, “errmsg”: “BSONObj size: 20112313 (0x132E3B9) is invalid. Size must be between 0 and 16793600(16MB) First element: _id: { _data: “82603CB5170000001B2B022C0100296E5A10042F63D0C3B388496786DFA01F1DD6E98B46645F69640064603C9CFEFA49CA266000FDBE0004” }”, “code”: 10334, “codeName”: “BSONObjectTooLarge”, “$gleStats”: {“lastOpTime”: {"$timestamp": {“t”: 0, “i”: 0}}, “electionId”: {"$oid": “000000000000000000000000”}}, “lastCommittedOpTime”: {"$timestamp": {“t”: 1614669333, “i”: 23}}, “$configServerState”: {“opTime”: {“ts”: {"$timestamp": {“t”: 1614669330, “i”: 28}}, “t”: {"$numberLong": “26477”}}}, “$clusterTime”: {“clusterTime”: {"$timestamp": {“t”: 1614669333, “i”: 23}}, “signature”: {“hash”: {"$binary": “kj05HLvxFsV7IuikAsH2v0phu98=”, “$type”: “00”}, “keyId”: {"$numberLong": “6888062584168120353”}}}} (com.mongodb.kafka.connect.source.MongoSourceTask)

After this error connector stalemated. Restart of a connector also fails, because can’t resume using this token, because of the same error.

Is there any chance to fix it? Because restart it with copyExisting as per documentation is not an option for us?

1 Like
2

I am facing the same issue, did you manage to fix ?

3

No, this particular message we failed to process due to limitations (visible in error logs), but we managed to recover connector and push it to work further. What we did was: Python script took CDC signature, processed messages one-by-one and thus located the faulty message, then skipped it, got the next CDC signature. (For us problem was due to we were sending full documents to Kafka, so switching to sending only change part, reduced message size thus allowing to skip it), after we received next CDC signature we wrote it into Kafka Offset topic, so the connector continued to work.

But then we decided not to use connector that receives full documents at all, so problem was not actual anymore.

But I believe if you are seeking for long term working solution - we do not have it, unfortunately.

1 Like
4

  1. you can modify change stream output using pipeline.
    see:https://www.mongodb.com/docs/v5.2/changeStreams/#modify-change-stream-output
    for example, using $unset to remove unused fields (eg: updateDescription).
    if using debezium, You can modify the configuration item cursor.pipeline. see: Debezium connector for MongoDB :: Debezium Documentation

  2. set cursor.oversize.handling.mode to split mode to split stream event to multiple events
    Change events exceeding the maximum BSON size will be split using the $changeStreamSplitLargeEvent aggregation. This option requires MongoDB 6.0.9 or newer .