[source-mongodb] Incremental sync error on specific collection - Saved offset is not valid

### Connector Name

source-mongodb

### Connector Version

1.5.12

### What step the error happened?

During the sync

### Relevant information

After a successful initial sync, incremental replication randomly fails after 1-10 successful ones with a
`Saved offset is not valid. Please reset the connection, and then increase oplog retention and/or increase sync frequency to prevent this from happening in the future`.

I manually checked the resume token from the logs which is invalid, but manually testing a resume token from a previous successful sync works (which seems to confirm that oplog is not truncated).

Reducing the replication frequency to 5 min (with oplog ~4days retention) does not fix.

Some other collections from the same cluster sync without issues.

**More info**
- MongoDB 7.0.15 (Atlas-managed) / ReplicaSet
- Collection:
```
Storage Size:
78.14MB
Logical Data Size:
217.94MB
Total Documents:
39641
Indexes Total Size:
14.07MB
```
- Oplog configuration
- Logs
[failed_sync.log](https://github.com/user-attachments/files/17746898/failed_sync.log)
[mongo_db.log](https://github.com/user-attachments/files/17746897/mongo_db.log)

cc @theyueli

### Relevant log output

```shell
2024-11-04 18:30:19 source > INFO main i.a.i.s.m.c.MongoDbDebeziumStateUtil(isValidResumeToken):124 Exception : Command failed with error 280 (ChangeStreamFatalError): 'PlanExecutor error during aggregation :: caused by :: cannot resume stream; the resume token was not found. {_data: "82672911AA000000042B042C0100296E5A1004111CD8A7C3AC4108BD652688BF9B4961463C6F7065726174696F6E54797065003C7570646174650046646F63756D656E744B65790046645F696400646607170ADC704B0B4CEF9E4F000004"}' on server foo-prod-shard-00-02.abcde.mongodb.net:27017. The full response is {"errorLabels": ["NonResumableChangeStreamError"], "ok": 0.0, "errmsg": "PlanExecutor error during aggregation :: caused by :: cannot resume stream; the resume token was not found. {_data: \"82672911AA000000042B042C0100296E5A1004111CD8A7C3AC4108BD652688BF9B4961463C6F7065726174696F6E54797065003C7570646174650046646F63756D656E744B65790046645F696400646607170ADC704B0B4CEF9E4F000004\"}", "code": 280, "codeName": "ChangeStreamFatalError", "$clusterTime": {"clusterTime": {"$timestamp": {"t": 1730745019, "i": 5}}, "signature": {"hash": {"$binary": {"base64": "8gooMnQDcDZ4N47zoh55j9GzBjk=", "subType": "00"}}, "keyId": 7428267533251117062}}, "operationTime": {"$timestamp": {"t": 1730745019, "i": 4}}}
2024-11-04 18:30:19 source > INFO main i.a.i.s.m.c.MongoDbDebeziumStateUtil(isValidResumeToken):125 Invalid resume token '826729119A0000003D2B042C0100296E5A100479290C5D090D4810962649B5F651BDB3463C6F7065726174696F6E54797065003C7570646174650046646F63756D656E744B65790046645F6964006466BDC93807F670F0E483042F000004' present, corresponding to timestamp (seconds after epoch) : 1730744730, due to reason Command failed with error 280 (ChangeStreamFatalError): 'PlanExecutor error during aggregation :: caused by :: cannot resume stream; the resume token was not found. {_data: "82672911AA000000042B042C0100296E5A1004111CD8A7C3AC4108BD652688BF9B4961463C6F7065726174696F6E54797065003C7570646174650046646F63756D656E744B65790046645F696400646607170ADC704B0B4CEF9E4F000004"}' on server foo-prod-shard-00-02.abcde.mongodb.net:27017. The full response is {"errorLabels": ["NonResumableChangeStreamError"], "ok": 0.0, "errmsg": "PlanExecutor error during aggregation :: caused by :: cannot resume stream; the resume token was not found. {_data: \"82672911AA000000042B042C0100296E5A1004111CD8A7C3AC4108BD652688BF9B4961463C6F7065726174696F6E54797065003C7570646174650046646F63756D656E744B65790046645F696400646607170ADC704B0B4CEF9E4F000004\"}", "code": 280, "codeName": "ChangeStreamFatalError", "$clusterTime": {"clusterTime": {"$timestamp": {"t": 1730745019, "i": 5}}, "signature": {"hash": {"$binary": {"base64": "8gooMnQDcDZ4N47zoh55j9GzBjk=", "subType": "00"}}, "keyId": 7428267533251117062}}, "operationTime": {"$timestamp": {"t": 1730745019, "i": 4}}}
2024-11-04 18:30:19 replication-orchestrator > SOURCE analytics [airbyte/source-mongodb-v2:1.5.12] | Type: db-sources-cdc-cursor-invalid | Value: 1
2024-11-04 18:30:19 source > ERROR main i.a.i.s.m.MongoDbSource(read):157 Unable to perform sync read operation. io.airbyte.commons.exceptions.ConfigErrorException: Saved offset is not valid. Please reset the connection, and then increase oplog retention and/or increase sync frequency to prevent his from happening in the future. See https://docs.airbyte.com/integrations/sources/mongodb-v2#mongodb-oplog-and-change-streams for more details
at io.airbyte.integrations.source.mongodb.cdc.MongoDbCdcInitializer.createCdcIterators(MongoDbCdcInitializer.java:136) ~[io.airbyte.airbyte-integrations.connectors-source-mongodb-v2.jar:?]
at io.airbyte.integrations.source.mongodb.MongoDbSource.read(MongoDbSource.java:149) [io.airbyte.airbyte-integrations.connectors-source-mongodb-v2.jar:?]
at io.airbyte.cdk.integrations.base.IntegrationRunner.readSerial(IntegrationRunner.kt:290) [airbyte-cdk-core-0.45.1.jar:?]
at io.airbyte.cdk.integrations.base.IntegrationRunner.runInternal(IntegrationRunner.kt:190) [airbyte-cdk-core-0.45.1.jar:?]
at io.airbyte.cdk.integrations.base.IntegrationRunner.run(IntegrationRunner.kt:119) [airbyte-cdk-core-0.45.1.jar:?]
at io.airbyte.cdk.integrations.base.IntegrationRunner.run$default(IntegrationRunner.kt:113) [airbyte-cdk-core-0.45.1.jar:?]
at io.airbyte.cdk.integrations.base.IntegrationRunner.run(IntegrationRunner.kt) [airbyte-cdk-core-0.45.1.jar:?]
at io.airbyte.integrations.source.mongodb.MongoDbSource.main(MongoDbSource.java:53) [io.airbyte.airbyte-integrations.connectors-source-mongodb-v2.jar:?]

Stack Trace: io.airbyte.commons.exceptions.ConfigErrorException: Saved offset is not valid. Please reset the connection, and then increase oplog retention and/or increase sync frequency to prevent his from happening in the future. See https://docs.airbyte.com/integrations/sources/mongodb-v2#mongodb-oplog-and-change-streams for more details
at io.airbyte.integrations.source.mongodb.cdc.MongoDbCdcInitializer.createCdcIterators(MongoDbCdcInitializer.java:136)
at io.airbyte.integrations.source.mongodb.MongoDbSource.read(MongoDbSource.java:149)
at io.airbyte.cdk.integrations.base.IntegrationRunner.readSerial(IntegrationRunner.kt:290)
at io.airbyte.cdk.integrations.base.IntegrationRunner.runInternal(IntegrationRunner.kt:190)
at io.airbyte.cdk.integrations.base.IntegrationRunner.run(IntegrationRunner.kt:119)
at io.airbyte.cdk.integrations.base.IntegrationRunner.run$default(IntegrationRunner.kt:113)
at io.airbyte.cdk.integrations.base.IntegrationRunner.run(IntegrationRunner.kt)
at io.airbyte.integrations.source.mongodb.MongoDbSource.main(MongoDbSource.java:53)

2024-11-04 18:30:19 source > ERROR main i.a.c.i.u.ConnectorExceptionHandler(handleException):68 caught exception! io.airbyte.commons.exceptions.ConfigErrorException: Saved offset is not valid. Please reset the connection, and then increase oplog retention and/or increase sync frequency to prevent his from happening in the future. See https://docs.airbyte.com/integrations/sources/mongodb-v2#mongodb-oplog-and-change-streams for more details
at io.airbyte.integrations.source.mongodb.cdc.MongoDbCdcInitializer.createCdcIterators(MongoDbCdcInitializer.java:136) ~[io.airbyte.airbyte-integrations.connectors-source-mongodb-v2.jar:?]
at io.airbyte.integrations.source.mongodb.MongoDbSource.read(MongoDbSource.java:149) ~[io.airbyte.airbyte-integrations.connectors-source-mongodb-v2.jar:?]
at io.airbyte.cdk.integrations.base.IntegrationRunner.readSerial(IntegrationRunner.kt:290) ~[airbyte-cdk-core-0.45.1.jar:?]
at io.airbyte.cdk.integrations.base.IntegrationRunner.runInternal(IntegrationRunner.kt:190) [airbyte-cdk-core-0.45.1.jar:?]
at io.airbyte.cdk.integrations.base.IntegrationRunner.run(IntegrationRunner.kt:119) [airbyte-cdk-core-0.45.1.jar:?]
at io.airbyte.cdk.integrations.base.IntegrationRunner.run$default(IntegrationRunner.kt:113) [airbyte-cdk-core-0.45.1.jar:?]
at io.airbyte.cdk.integrations.base.IntegrationRunner.run(IntegrationRunner.kt) [airbyte-cdk-core-0.45.1.jar:?]
at io.airbyte.integrations.source.mongodb.MongoDbSource.main(MongoDbSource.java:53) [io.airbyte.airbyte-integrations.connectors-source-mongodb-v2.jar:?]

Stack Trace: io.airbyte.commons.exceptions.ConfigErrorException: Saved offset is not valid. Please reset the connection, and then increase oplog retention and/or increase sync frequency to prevent his from happening in the future. See https://docs.airbyte.com/integrations/sources/mongodb-v2#mongodb-oplog-and-change-streams for more details
at io.airbyte.integrations.source.mongodb.cdc.MongoDbCdcInitializer.createCdcIterators(MongoDbCdcInitializer.java:136)
at io.airbyte.integrations.source.mongodb.MongoDbSource.read(MongoDbSource.java:149)
at io.airbyte.cdk.integrations.base.IntegrationRunner.readSerial(IntegrationRunner.kt:290)
at io.airbyte.cdk.integrations.base.IntegrationRunner.runInternal(IntegrationRunner.kt:190)
at io.airbyte.cdk.integrations.base.IntegrationRunner.run(IntegrationRunner.kt:119)
at io.airbyte.cdk.integrations.base.IntegrationRunner.run$default(IntegrationRunner.kt:113)
at io.airbyte.cdk.integrations.base.IntegrationRunner.run(IntegrationRunner.kt)
at io.airbyte.integrations.source.mongodb.MongoDbSource.main(MongoDbSource.java:53)

2024-11-04 18:30:19 replication-orchestrator > readFromSource: source exception
io.airbyte.workers.internal.exception.SourceException: Source process exited with non-zero exit code 1
at io.airbyte.workers.general.BufferedReplicationWorker.readFromSource(BufferedReplicationWorker.java:364) ~[io.airbyte-airbyte-commons-worker-1.1.0.jar:?]
at io.airbyte.workers.general.BufferedReplicationWorker.lambda$runAsyncWithHeartbeatCheck$3(BufferedReplicationWorker.java:222) ~[io.airbyte-airbyte-commons-worker-1.1.0.jar:?]
at java.base/java.util.concurrent.CompletableFuture$AsyncRun.run(CompletableFuture.java:1804) ~[?:?]
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144) ~[?:?]
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642) ~[?:?]
at java.base/java.lang.Thread.run(Thread.java:1583) [?:?]
```

### Contribute

- [ ] Yes, I want to contribute