Resume token has expired, fallback to timestamp restart mode

Error message

Resume token has expired, fallback to timestamp restart mode

What it means

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.

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:134

        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;
                    }

                    if (heartbeatManager != null) {

View on GitHub (pinned to cf67b549a7)

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

Example fix

// connector-internal fallback (already implemented)
boolean resumeTokenExpires = MongodbUtils.checkIfResumeTokenExpires(e);
if (resumeTokenExpires) {
    log.warn("Resume token has expired, fallback to timestamp restart mode");
}
changeStreamCursor = openChangeStreamCursor(descriptor, resumeTokenExpires);
// user-side mitigation: enlarge oplog on the MongoDB replica set
use local
db.runCommand({ replSetResizeOplog: 1, size: 20480 }) // 20 GB

Defensive patterns

Strategy: fallback

Validate before calling

// Before resuming from a stored token, verify the token is still resumable
boolean tokenStillValid(BsonDocument resumeToken) {
    try {
        collection.watch().resumeAfter(resumeToken).cursor().close();
        return true;
    } catch (MongoCommandException e) {
        return e.getErrorCode() != 9 /* FailedToParse */ && !isTokenExpired(e);
    }
}

Try / catch

try {
    next = Optional.ofNullable(changeStreamCursor.tryNext());
} catch (MongoCommandException e) {
    if (MongodbUtils.checkIfResumeTokenExpires(e)) {
        // accept at-least-once re-read from timestamp fallback
        changeStreamCursor = openChangeStreamCursor(descriptor, true);
    } else {
        throw e;
    }
}

Prevention

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

When it happens

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

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

Related errors

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