Hi Sonal, Thank you for the detailed writeup.
If I understand correctly, StreamSplitState will be updated in MongoDBRecordEmitter, so the resume token and starting timestamp should be updated in-place. Could you please file a minimal reproducible example to verify this bug? The error log and large BSONObject issue seems irrelevant to the state recovery issue, maybe we should create individual JIRA tickets / PRs to trace and fix them. Looking forward to your patch. Best Regards, Xiqian > Sonal Maheshwari <[email protected]> 於 2026年10月19日 下午2:30 寫道： > > 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/
