Change stream cursor has expired, trying to recreate cursor

Error message

Change stream cursor has expired, trying to recreate cursor

What it means

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.

Source

Thrown at seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/source/fetch/MongodbStreamFetchTask.java:131

        ChangeStreamDescriptor descriptor = taskContext.getChangeStreamDescriptor();
        ChangeEventQueue<DataChangeEvent> queue = taskContext.getQueue();

        this.mongoClient = taskContext.getMongoClient();
        MongoChangeStreamCursor<BsonDocument> changeStreamCursor =
                openChangeStreamCursor(descriptor);
        HeartbeatManager heartbeatManager = openHeartbeatManagerIfNeeded(changeStreamCursor);

        final long startPoll = time.milliseconds();
        long nextUpdate = startPoll + sourceConfig.getPollAwaitTimeMillis();
        this.taskRunning = true;
        try {
            while (taskRunning) {
                Optional<BsonDocument> next;
                try {
                    next = Optional.ofNullable(changeStreamCursor.tryNext());
                } catch (MongoCommandException e) {
                    if (MongodbUtils.checkIfChangeStreamCursorExpires(e)) {
                        log.warn("Change stream cursor has expired, trying to recreate cursor");
                        boolean resumeTokenExpires = MongodbUtils.checkIfResumeTokenExpires(e);
                        if (resumeTokenExpires) {
                            log.warn(
                                    "Resume token has expired, fallback to timestamp restart mode");
                        }
                        changeStreamCursor = openChangeStreamCursor(descriptor, resumeTokenExpires);
                        next = Optional.ofNullable(changeStreamCursor.tryNext());
                    } else {
                        throw e;
                    }
                }
                SourceRecord changeRecord = null;
                if (!next.isPresent()) {
                    long untilNext = nextUpdate - time.milliseconds();
                    if (untilNext > 0) {
                        log.debug("Waiting {} ms to poll change records", untilNext);
                        time.sleep(untilNext);
                        continue;

View on GitHub (pinned to cf67b549a7)

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

Example fix

// connector-internal recovery (already implemented)
catch (MongoCommandException e) {
    if (MongodbUtils.checkIfChangeStreamCursorExpires(e)) {
        changeStreamCursor = openChangeStreamCursor(descriptor, resumeTokenExpires);
    } else { throw e; }
}
// user-side mitigation: enable heartbeat in connector config to keep tokens fresh
"heartbeat.interval.ms" = "30000"

Defensive patterns

Strategy: retry

Validate before calling

// Health probe before/alongside the job: ensure the change stream can be opened
try (MongoChangeStreamCursor<BsonDocument> c = collection.watch()
        .cursor()) {
    BsonDocument first = c.tryNext(); // exercises cursor establishment
}

Try / catch

try {
    next = Optional.ofNullable(changeStreamCursor.tryNext());
} catch (MongoCommandException e) {
    if (e.getErrorCode() == 237 || e.getErrorCode() == 43) { // CursorKilled / CursorNotFound
        changeStreamCursor = openChangeStreamCursor(descriptor, false); // bounded retry
    } else {
        throw e;
    }
}

Prevention

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

When it happens

Trigger: 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.

Common situations: 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.

Related errors

AI-assisted analysis of apache/seatunnel@cf67b549a7 (2026-09-10). Data as JSON: /api/errors/076d1611364e59d7. Report an issue: GitHub.