Hi all, I've found a bug in flink-connector-mongodb-cdc's error recovery and would like to contribute a fix. (JIRA account requested; I'll file a ticket once approved.)
When tryNext() fails with a code in INVALID_CHANGE_STREAM_ERRORS, the fetch task recreates the cursor in place [1], but builds the resume position from streamSplit.getStartingOffset() [2] -- which is set when the task starts and never advances. So the cursor reopens from task start and replays everything read since. In production this gave us a multi-day lag spike from one error. For 280/286 it's worse: the timestamp fallback [3] also comes from the starting offset, so after "resume token too old" it retries from an even older point. The error itself is also invisible: the only log line is a fixed WARN, "Change stream cursor has expired, trying to recreate cursor" [4], with no error code or message. With six codes going down this path, there's no way to tell from logs which one fired. We only found ours by patching that line to log the exception. Our trigger turned out to be BSONObjectTooLarge (10334): with updateLookup, an update event can exceed 16MB even when the document doesn't. It's in the recreate list [5], but recreating can't help since the event itself is too large. Proposed: 1. Track the latest resume token (events + heartbeats) and reopen from it. Restart/restore behaviour is unchanged. Also log the error code and message. 2. Separately, use $changeStreamSplitLargeEvent (MongoDB 6.0.9+) [6] and reassemble fragments in the fetch task. I have patches for both. Feedback welcome. Thanks, Sonal [1] https://github.com/apache/flink-cdc/blob/abf18d7a87e6937984949901b77b87ce6def5a76/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/reader/fetch/MongoDBStreamFetchTask.java#L115-L127 [2] https://github.com/apache/flink-cdc/blob/abf18d7a87e6937984949901b77b87ce6def5a76/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/reader/fetch/MongoDBStreamFetchTask.java#L258-L259 [3] https://github.com/apache/flink-cdc/blob/abf18d7a87e6937984949901b77b87ce6def5a76/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/reader/fetch/MongoDBStreamFetchTask.java#L279 [4] https://github.com/apache/flink-cdc/blob/abf18d7a87e6937984949901b77b87ce6def5a76/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/reader/fetch/MongoDBStreamFetchTask.java#L117 [5] https://github.com/apache/flink-cdc/blob/abf18d7a87e6937984949901b77b87ce6def5a76/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/utils/MongoUtils.java#L90-L98 [6] https://www.mongodb.com/docs/manual/reference/operator/aggregation/changeStreamSplitLargeEvent/
