{"record":{"id":"feb9412a8ffb841a","repo":"apache/seatunnel","slug":"resume-token-has-expired-fallback-to-timestamp-re","errorCode":null,"errorMessage":"Resume token has expired, fallback to timestamp restart mode","messagePattern":"Resume token has expired, fallback to timestamp restart mode","errorType":"console","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/source/fetch/MongodbStreamFetchTask.java","lineNumber":134,"sourceCode":"        this.mongoClient = taskContext.getMongoClient();\n        MongoChangeStreamCursor<BsonDocument> changeStreamCursor =\n                openChangeStreamCursor(descriptor);\n        HeartbeatManager heartbeatManager = openHeartbeatManagerIfNeeded(changeStreamCursor);\n\n        final long startPoll = time.milliseconds();\n        long nextUpdate = startPoll + sourceConfig.getPollAwaitTimeMillis();\n        this.taskRunning = true;\n        try {\n            while (taskRunning) {\n                Optional<BsonDocument> next;\n                try {\n                    next = Optional.ofNullable(changeStreamCursor.tryNext());\n                } catch (MongoCommandException e) {\n                    if (MongodbUtils.checkIfChangeStreamCursorExpires(e)) {\n                        log.warn(\"Change stream cursor has expired, trying to recreate cursor\");\n                        boolean resumeTokenExpires = MongodbUtils.checkIfResumeTokenExpires(e);\n                        if (resumeTokenExpires) {\n                            log.warn(\n                                    \"Resume token has expired, fallback to timestamp restart mode\");\n                        }\n                        changeStreamCursor = openChangeStreamCursor(descriptor, resumeTokenExpires);\n                        next = Optional.ofNullable(changeStreamCursor.tryNext());\n                    } else {\n                        throw e;\n                    }\n                }\n                SourceRecord changeRecord = null;\n                if (!next.isPresent()) {\n                    long untilNext = nextUpdate - time.milliseconds();\n                    if (untilNext > 0) {\n                        log.debug(\"Waiting {} ms to poll change records\", untilNext);\n                        time.sleep(untilNext);\n                        continue;\n                    }\n\n                    if (heartbeatManager != null) {","sourceCodeStart":116,"sourceCodeEnd":152,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/source/fetch/MongodbStreamFetchTask.java#L116-L152","documentation":"A sub-case of cursor expiry in MongodbStreamFetchTask.execute: when the expired change stream cursor cannot be resumed because the stored resume token is no longer valid on the server (the oplog entry it points to has been rolled off), checkIfResumeTokenExpires(e) returns true and the task logs this warning before falling back to timestamp restart mode. Instead of resuming, it re-opens the change stream from a computed timestamp, which can cause re-reading (at-least-once semantics) but avoids unrecoverable failure.","triggerScenarios":"execute()'s tryNext() throws MongoCommandException where both checkIfChangeStreamCursorExpires and checkIfResumeTokenExpires hold — the resume token references an oplog position no longer retained, after long connector downtime, heavy write volume flushing the oplog, or a MongoDB upgrade that invalidates tokens.","commonSituations":"Connector paused/offline for longer than the oplog retention window while the source collection stays busy; small oplog size (default) on busy clusters causing rapid rollover; version upgrades of MongoDB altering resume-token semantics; checkpoint restore from a very old offset.","solutions":["No immediate failure — the task falls back to timestamp restart; be aware downstream may receive duplicate events (at-least-once), so sinks must be idempotent","Increase MongoDB oplog size (oplogSizeMB) so the retained window comfortably exceeds potential connector downtime","Reduce downtime: keep the connector running or restart it promptly; restore from recent checkpoints rather than stale offsets","Enable connector heartbeats so offsets/resume tokens advance even during quiet periods"],"exampleFix":"// connector-internal fallback (already implemented)\nboolean resumeTokenExpires = MongodbUtils.checkIfResumeTokenExpires(e);\nif (resumeTokenExpires) {\n    log.warn(\"Resume token has expired, fallback to timestamp restart mode\");\n}\nchangeStreamCursor = openChangeStreamCursor(descriptor, resumeTokenExpires);\n// user-side mitigation: enlarge oplog on the MongoDB replica set\nuse local\ndb.runCommand({ replSetResizeOplog: 1, size: 20480 }) // 20 GB","handlingStrategy":"fallback","validationCode":"// Before resuming from a stored token, verify the token is still resumable\nboolean tokenStillValid(BsonDocument resumeToken) {\n    try {\n        collection.watch().resumeAfter(resumeToken).cursor().close();\n        return true;\n    } catch (MongoCommandException e) {\n        return e.getErrorCode() != 9 /* FailedToParse */ && !isTokenExpired(e);\n    }\n}","typeGuard":null,"tryCatchPattern":"try {\n    next = Optional.ofNullable(changeStreamCursor.tryNext());\n} catch (MongoCommandException e) {\n    if (MongodbUtils.checkIfResumeTokenExpires(e)) {\n        // accept at-least-once re-read from timestamp fallback\n        changeStreamCursor = openChangeStreamCursor(descriptor, true);\n    } else {\n        throw e;\n    }\n}","preventionTips":["Size the oplog (oplogSizeMB) to cover your maximum tolerable connector downtime on busy clusters","Make downstream sinks idempotent/upsert-capable since timestamp fallback can replay events","Keep the connector running continuously and restore only from recent checkpoints, never stale offsets","Alert on 'Resume token has expired' warnings — each one signals data re-read beyond the oplog window"],"tags":["mongodb","cdc","change-stream","resume-token","oplog"],"backgroundTag":"cursor-expired","analyzedSha":"cf67b549a7a6c35fa0beb12d83c62892427ea919","analyzedAt":"2026-09-10T21:44:55.265Z","contentChangedAt":"2026-09-10T21:44:55.265Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}