{"record":{"id":"076d1611364e59d7","repo":"apache/seatunnel","slug":"change-stream-cursor-has-expired-trying-to-recrea","errorCode":null,"errorMessage":"Change stream cursor has expired, trying to recreate cursor","messagePattern":"Change stream cursor has expired, trying to recreate cursor","errorType":"console","errorClass":null,"httpStatus":null,"severity":"info","filePath":"seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/source/fetch/MongodbStreamFetchTask.java","lineNumber":131,"sourceCode":"        ChangeStreamDescriptor descriptor = taskContext.getChangeStreamDescriptor();\n        ChangeEventQueue<DataChangeEvent> queue = taskContext.getQueue();\n\n        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;","sourceCodeStart":113,"sourceCodeEnd":149,"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#L113-L149","documentation":"MongodbStreamFetchTask.execute polls a MongoDB change stream via changeStreamCursor.tryNext() inside its main run loop. When tryNext() throws a MongoCommandException whose error code indicates the server-side cursor has expired, the task logs this warning and recreates the cursor instead of failing. This is an expected, recoverable condition: MongoDB change stream cursors have a server lifetime and can be re-opened from the last resume token.","triggerScenarios":"execute()'s loop calls changeStreamCursor.tryNext(); MongoDB replies with a command error (e.g. code 237 CursorKilled / 43 CursorNotFound, errorLabels indicating resume) that MongodbUtils.checkIfChangeStreamCursorExpires(e) classifies as cursor expiry — typically after long idle periods, server restarts, failovers, or cursor reaping by the deployment.","commonSituations":"Low-traffic collections with no change events for longer than the cursor idle timeout; MongoDB primary failover or replica set election invalidating cursors; long-running CDC jobs spanning maintenance windows; server-side cursor lifetime limits reached during large backlogs.","solutions":["No user action strictly required — the task recreates the cursor and continues; ensure the stream task's retry/reopen path is active","If expiry is frequent, tune MongoDB cursor/heartbeat settings: enable the connector's heartbeat so resume tokens refresh during quiet periods","Keep the resume token persisted (offset) so cursor recreation resumes without re-snapshotting","Check replica set health and idle timeouts if recreations happen excessively, since each recreation adds latency"],"exampleFix":"// connector-internal recovery (already implemented)\ncatch (MongoCommandException e) {\n    if (MongodbUtils.checkIfChangeStreamCursorExpires(e)) {\n        changeStreamCursor = openChangeStreamCursor(descriptor, resumeTokenExpires);\n    } else { throw e; }\n}\n// user-side mitigation: enable heartbeat in connector config to keep tokens fresh\n\"heartbeat.interval.ms\" = \"30000\"","handlingStrategy":"retry","validationCode":"// Health probe before/alongside the job: ensure the change stream can be opened\ntry (MongoChangeStreamCursor<BsonDocument> c = collection.watch()\n        .cursor()) {\n    BsonDocument first = c.tryNext(); // exercises cursor establishment\n}","typeGuard":null,"tryCatchPattern":"try {\n    next = Optional.ofNullable(changeStreamCursor.tryNext());\n} catch (MongoCommandException e) {\n    if (e.getErrorCode() == 237 || e.getErrorCode() == 43) { // CursorKilled / CursorNotFound\n        changeStreamCursor = openChangeStreamCursor(descriptor, false); // bounded retry\n    } else {\n        throw e;\n    }\n}","preventionTips":["Enable a short heartbeat interval so resume tokens stay fresh during quiet periods","Persist offsets/resume tokens so automatic cursor recreation never forces a re-snapshot","Monitor logs for recurring cursor-expiry warnings; frequent occurrences indicate failover or timeout tuning needs","Ensure replica set health and adequate cursor/idle timeouts for low-traffic collections"],"tags":["mongodb","cdc","change-stream","cursor","retry"],"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"}