Why are we looking for errorMessage.contains(RESUME_TOKEN)

[sjmittal]

@yuxiqian In the following code, why do we check for errorMessage.contains(RESUME_TOKEN) ?

https://github.com/apache/flink-cdc/blob/master/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/utils/MongoUtils.java#L462

For errors messages like:
com.mongodb.MongoCommandException: Command failed with error 286 (ChangeStreamHistoryLost): 'PlanExecutor error during aggregation :: caused by :: Resume of change stream was not possible, as the resume point may no longer be in the oplog.' on xxxxx:27017. The full response is {"errorLabels": ["NonResumableChangeStreamError"], "ok": 0.0, "errmsg": "PlanExecutor error during aggregation :: caused by :: Resume of change stream was not possible, as the resume point may no longer be in the oplog.", "code": 286, "codeName": "ChangeStreamHistoryLost", "$clusterTime": {"clusterTime": {"$timestamp": {"t": 1760708556, "i": 4}}, "signature": {"hash": {"$binary": {"base64": "TbHFeBl/kiI3w3EG+TjtkkNn5Sk=", "subType": "00"}}, "keyId": 7511389671713144834}}, "operationTime": {"$timestamp": {"t": 1760708556, "i": 4}}}
at com.mongodb.internal.connection.ProtocolHelper.getCommandFailureException(ProtocolHelper.java:205)
at com.mongodb.internal.connection.InternalStreamConnection.receiveCommandMessageResponse(InternalStreamConnection.java:454)
at com.mongodb.internal.connection.InternalStreamConnection.sendAndReceive(InternalStreamConnection.java:372)
at com.mongodb.internal.connection.UsageTrackingInternalConnection.sendAndReceive(UsageTrackingInternalConnection.java:114)
at com.mongodb.internal.connection.DefaultConnectionPool$PooledConnection.sendAndReceive(DefaultConnectionPool.java:765)
at com.mongodb.internal.connection.CommandProtocolImpl.execute(CommandProtocolImpl.java:76)
at com.mongodb.internal.connection.DefaultServer$DefaultServerProtocolExecutor.execute(DefaultServer.java:209)
at com.mongodb.internal.connection.DefaultServerConnection.executeProtocol(DefaultServerConnection.java:115)
at com.mongodb.internal.connection.DefaultServerConnection.command(DefaultServerConnection.java:83)
at com.mongodb.internal.connection.DefaultServerConnection.command(DefaultServerConnection.java:74)
at com.mongodb.internal.connection.DefaultServer$OperationCountTrackingConnection.command(DefaultServer.java:299)
at com.mongodb.internal.operation.SyncOperationHelper.createReadCommandAndExecute(SyncOperationHelper.java:273)
at com.mongodb.internal.operation.SyncOperationHelper.lambda$executeRetryableRead$3(SyncOperationHelper.java:191)
at com.mongodb.internal.operation.SyncOperationHelper.lambda$withSourceAndConnection$0(SyncOperationHelper.java:127)
at com.mongodb.internal.operation.SyncOperationHelper.withSuppliedResource(SyncOperationHelper.java:152)
at com.mongodb.internal.operation.SyncOperationHelper.lambda$withSourceAndConnection$1(SyncOperationHelper.java:126)
at com.mongodb.internal.operation.SyncOperationHelper.withSuppliedResource(SyncOperationHelper.java:152)
at com.mongodb.internal.operation.SyncOperationHelper.withSourceAndConnection(SyncOperationHelper.java:125)
at com.mongodb.internal.operation.SyncOperationHelper.lambda$executeRetryableRead$4(SyncOperationHelper.java:189)
at com.mongodb.internal.operation.SyncOperationHelper.lambda$decorateReadWithRetries$12(SyncOperationHelper.java:292)
at com.mongodb.internal.async.function.RetryingSyncSupplier.get(RetryingSyncSupplier.java:67)
at com.mongodb.internal.operation.SyncOperationHelper.executeRetryableRead(SyncOperationHelper.java:194)
at com.mongodb.internal.operation.SyncOperationHelper.executeRetryableRead(SyncOperationHelper.java:176)
at com.mongodb.internal.operation.AggregateOperationImpl.execute(AggregateOperationImpl.java:193)
at com.mongodb.internal.operation.ChangeStreamOperation.lambda$execute$0(ChangeStreamOperation.java:187)
at com.mongodb.internal.operation.SyncOperationHelper.withReadConnectionSource(SyncOperationHelper.java:99)
at com.mongodb.internal.operation.ChangeStreamOperation.execute(ChangeStreamOperation.java:185)
at com.mongodb.internal.operation.ChangeStreamOperation.execute(ChangeStreamOperation.java:54)
at com.mongodb.client.internal.MongoClientDelegate$DelegateOperationExecutor.execute(MongoClientDelegate.java:153)
at com.mongodb.client.internal.ChangeStreamIterableImpl.execute(ChangeStreamIterableImpl.java:212)
at com.mongodb.client.internal.ChangeStreamIterableImpl.access$000(ChangeStreamIterableImpl.java:55)
at com.mongodb.client.internal.ChangeStreamIterableImpl$1.cursor(ChangeStreamIterableImpl.java:139)
at com.mongodb.client.internal.ChangeStreamIterableImpl$1.cursor(ChangeStreamIterableImpl.java:131)
at org.apache.flink.cdc.connectors.mongodb.source.reader.fetch.MongoDBStreamFetchTask.openChangeStreamCursor(MongoDBStreamFetchTask.java:287)
at org.apache.flink.cdc.connectors.mongodb.source.reader.fetch.MongoDBStreamFetchTask.openChangeStreamCursor(MongoDBStreamFetchTask.java:249)
at org.apache.flink.cdc.connectors.mongodb.source.reader.fetch.MongoDBStreamFetchTask.execute(MongoDBStreamFetchTask.java:104)
at org.apache.flink.cdc.connectors.base.source.reader.external.IncrementalSourceStreamFetcher.lambda$submitTask$0(IncrementalSourceStreamFetcher.java:93)
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
at java.base/java.lang.Thread.run(Thread.java:829)

It simply fails because of that added check. If we make the code like this:

return errorMessage.contains(RESUME_TOKEN)
|| errorMessage.contains(NOT_FOUND)
|| errorMessage.contains(DOES_NOT_EXIST)
|| errorMessage.contains(INVALID_RESUME_TOKEN)
|| errorMessage.contains(NO_LONGER_IN_THE_OPLOG);

Then it will cover this case also.

What do you think ?

[yuxiqian]

Hello @sjmittal, thanks for your detailed writeup.

One thing I'm not quite sure about is if concatenating errorMessage.contains(RESUME_TOKEN) with || is appropriate, since "resume token" might occur in many other exception messages, some of which we don't want to capture.

Agree this code needs to be fixed, perhaps by matching full error messages instead of keywords. Would you like to take this?

[sjmittal]

@yuxiqian The list of mongodb error codes are described in here:
https://www.mongodb.com/docs/manual/reference/error-codes/

However full list of actual error messages are embedded all over in the mongodb code and these error texts are subjected change.

Looking at the MongoDB CDC code I see that all the needed error codes are captured for INVALID_CHANGE_STREAM_ERRORS.

If we modify the method checkIfResumeTokenExpires like this, we should cover all the case and direct the CDC to startAtOperationTime.

public static boolean checkIfResumeTokenExpires(final MongoCommandException e) {
        if (e.getCode() != CHANGE_STREAM_FATAL_ERROR) {
            return false;
        }
        String errorMessage = e.getErrorMessage().toLowerCase(Locale.ROOT);
        return errorMessage.contains(RESUME_TOKEN)
                    || errorMessage.contains(NOT_FOUND)
                    || errorMessage.contains(DOES_NOT_EXIST)
                    || errorMessage.contains(NO_LONGER_IN_THE_OPLOG);
    }

Let me know what do you think ?

[yuxiqian]

You're right, it's fragile to rely on error messages as they're not documented.

For the code snippet here, should we also modify the if (e.getCode() != CHANGE_STREAM_FATAL_ERROR) check? It seems the MongoCommandException above is reporting a CHANGE_STREAM_HISTORY_LOST (286) error code instead of CHANGE_STREAM_FATAL_ERROR (280).

[sjmittal]

@yuxiqian I have created a JIRA for this: https://issues.apache.org/jira/browse/FLINK-38550
I have also created a PR for this: #4161

I think this would be better to review the change. I have incorporated your suggestion.

Let me know if this is good to merge.

[yuxiqian]

Thanks! Let's move this discussion there.