[Bug] [MONGODB-CDC] MongoDB CDC snapshot read fails with “Illegal argument” in streaming job

### Search before asking

- [x] I had searched in the [issues](https://github.com/apache/seatunnel/issues?q=is%3Aissue+label%3A%22bug%22) and found no similar issues.

### What happened

I am trying to run a streaming job to replicate MongoDB CDC data into a MySQL database using `SeaTunnel 2.3.12` with Docker. Expected behavior is that All CDC data from MongoDB playground.tasks collection should be streamed into MySQL public.tasks_cdc table without errors, and the job statistics should reflect the number of records read and written. but When using the default startup.mode = 'INITIAL', the job fails during the snapshot read with the following error:

`org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.exception.MongodbConnectorException: ErrorCode:[COMMON-06], ErrorDescription:[Illegal argument] - Execute snapshot read subtask for mongodb split SnapshotSplit(tableId=playground.tasks, splitKeyType=ROW<_id INT>, splitStart=[{"_id": 1}, {"_id": {"$minKey": 1}}], splitEnd=[{"_id": 1}, {"_id": {"$maxKey": 1}}], lowWatermark=null, highWatermark=null) fail`

Switching startup.mode to 'TIMESTAMP' avoids this snapshot error, but the CDC data is not actually processed. Logs show “Message Delayed” and the job statistics report 0 records read and written, even though CDC events are occurring in MongoDB.

### SeaTunnel Version

2.3.12

### SeaTunnel Config

```conf
{
"env" : {
"parallelism" : 1,
"job.mode" : "STREAMING",
"checkpoint.interval" : 6000,
"read_limit.bytes_per_second" : 7000000,
"read_limit.rows_per_second" : 400
},
"source" : [
{
"plugin_output" : "mongocdc",
"hosts" : "xxx:27017,xxx:27017,xxx:27017",
"username" : "******",
"password" : "******",
"connection.options" : "replicaSet=xxxx&authSource=admin&tls=true&retryWrites=true&w=majority&appName=ChangeStream",
"database" : [
"playground"
],
"collection" : [
"playground.tasks"
],
"schema" : {
"columns" : [
{
"name" : "_id",
"type" : "string",
"nullable" : false,
"columnLength" : 255
},
{
"name" : "value",
"type" : "string"
}
]
},
"plugin_name" : "MongoDB-CDC"
}
],
"transform" : [
{
"plugin_input" : "mongocdc",
"plugin_output" : "parsed_cdc",
"columns" : [
{
"src_field" : "value",
"path" : "$.value._id._id._id.$oid",
"dest_field" : "_id"
},
{
"src_field" : "value",
"path" : "$.value.fullDocument",
"dest_field" : "fullDocument"
},
{
"src_field" : "value",
"path" : "$.value.documentKey._id.$oid",
"dest_field" : "documentKey"
},
{
"src_field" : "value",
"path" : "$.value.clusterTime.$timestamp.t",
"dest_field" : "loadTime"
},
{
"src_field" : "value",
"path" : "$.value.operationType",
"dest_field" : "operationType"
}
],
"plugin_name" : "JsonPath"
},
{
"plugin_input" : "parsed_cdc",
"plugin_output" : "final_transform",
"query" : "\r\n select \r\n _id as _id,\r\n case \r\n when operationType = 'insert' then 'I'\r\n when operationType = 'update' then 'U'\r\n when operationType = 'delete' then 'D'\r\n END as aws_dms_operation,\r\n fullDocument as _doc,\r\n loadTime as load_time\r\n from parsed_cdc\r\n ",
"plugin_name" : "Sql"
}
],
"sink" : [
{
"driver" : "com.mysql.cj.jdbc.Driver",
"url" : "jdbc:mysql://mysql0:3306/stdb?useSSL=false&allowPublicKeyRetrieval=true&serverTimezone=UTC&useUnicode=true&characterEncoding=UTF-8&rewriteBatchedStatements=true",
"username" : "******",
"password" : "******",
"plugin_input" : "final_transform",
"generate_sink_sql" : true,
"table" : "public.tasks_cdc",
"database" : "stdb",
"primary_keys" : [
"_id"
],
"schema_save_mode" : "CREATE_SCHEMA_WHEN_NOT_EXIST",
"data_save_mode" : "APPEND_DATA",
"plugin_name" : "JDBC"
}
]
}
```

### Running Command

```shell
docker run --rm -it --network=custom_network -v /seatunnel/config/:/config apache/seatunnel:2.3.12 ./bin/seatunnel.sh -m local -c /config/mongo_cdc-stream.conf
```

### Error Exception

```log
Exception in thread "main" org.apache.seatunnel.core.starter.exception.CommandExecuteException: SeaTunnel job executed failed
at org.apache.seatunnel.core.starter.seatunnel.command.ClientExecuteCommand.execute(ClientExecuteCommand.java:228)
at org.apache.seatunnel.core.starter.SeaTunnel.run(SeaTunnel.java:40)
at org.apache.seatunnel.core.starter.seatunnel.SeaTunnelClient.main(SeaTunnelClient.java:40)
Caused by: org.apache.seatunnel.engine.common.exception.SeaTunnelEngineException: java.lang.RuntimeException: One or more fetchers have encountered exception
at org.apache.seatunnel.connectors.seatunnel.common.source.reader.fetcher.SplitFetcherManager.checkErrors(SplitFetcherManager.java:147)
at org.apache.seatunnel.connectors.seatunnel.common.source.reader.SourceReaderBase.getNextFetch(SourceReaderBase.java:167)
at org.apache.seatunnel.connectors.seatunnel.common.source.reader.SourceReaderBase.pollNext(SourceReaderBase.java:93)
at org.apache.seatunnel.connectors.cdc.base.source.reader.IncrementalSourceReader.pollNext(IncrementalSourceReader.java:119)
at org.apache.seatunnel.engine.server.task.flow.SourceFlowLifeCycle.collect(SourceFlowLifeCycle.java:159)
at org.apache.seatunnel.engine.server.task.SourceSeaTunnelTask.collect(SourceSeaTunnelTask.java:127)
at org.apache.seatunnel.engine.server.task.SeaTunnelTask.stateProcess(SeaTunnelTask.java:165)
at org.apache.seatunnel.engine.server.task.SourceSeaTunnelTask.call(SourceSeaTunnelTask.java:132)
at org.apache.seatunnel.engine.server.TaskExecutionService$BlockingWorker.run(TaskExecutionService.java:679)
at org.apache.seatunnel.engine.server.TaskExecutionService$NamedTaskWrapper.run(TaskExecutionService.java:1008)
at org.apache.seatunnel.api.tracing.MDCRunnable.run(MDCRunnable.java:43)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:750)
Caused by: java.lang.RuntimeException: SplitFetcher thread 0 received unexpected exception while polling the records
at org.apache.seatunnel.connectors.seatunnel.common.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:165)
at org.apache.seatunnel.connectors.seatunnel.common.source.reader.fetcher.SplitFetcher.run(SplitFetcher.java:81)
... 5 more
Caused by: java.io.IOException: Source fetch execution was fail
at org.apache.seatunnel.connectors.seatunnel.common.source.reader.fetcher.FetchTask.run(FetchTask.java:73)
at org.apache.seatunnel.connectors.seatunnel.common.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:162)
... 6 more
Caused by: java.io.IOException: org.apache.seatunnel.common.utils.SeaTunnelException: Read split SnapshotSplit(tableId=playground.tasks, splitKeyType=ROW<_id INT>, splitStart=[{"_id": 1}, {"_id": {"$minKey": 1}}], splitEnd=[{"_id": 1}, {"_id": {"$maxKey": 1}}], lowWatermark=null, highWatermark=null) error due to ErrorCode:[COMMON-06], ErrorDescription:[Illegal argument] - Execute snapshot read subtask for mongodb split SnapshotSplit(tableId=playground.tasks, splitKeyType=ROW<_id INT>, splitStart=[{"_id": 1}, {"_id": {"$minKey": 1}}], splitEnd=[{"_id": 1}, {"_id": {"$maxKey": 1}}], lowWatermark=null, highWatermark=null) fail.
at org.apache.seatunnel.connectors.cdc.base.source.reader.IncrementalSourceSplitReader.fetch(IncrementalSourceSplitReader.java:78)
at org.apache.seatunnel.connectors.seatunnel.common.source.reader.fetcher.FetchTask.run(FetchTask.java:54)
... 7 more
Caused by: org.apache.seatunnel.common.utils.SeaTunnelException: Read split SnapshotSplit(tableId=playground.tasks, splitKeyType=ROW<_id INT>, splitStart=[{"_id": 1}, {"_id": {"$minKey": 1}}], splitEnd=[{"_id": 1}, {"_id": {"$maxKey": 1}}], lowWatermark=null, highWatermark=null) error due to ErrorCode:[COMMON-06], ErrorDescription:[Illegal argument] - Execute snapshot read subtask for mongodb split SnapshotSplit(tableId=playground.tasks, splitKeyType=ROW<_id INT>, splitStart=[{"_id": 1}, {"_id": {"$minKey": 1}}], splitEnd=[{"_id": 1}, {"_id": {"$maxKey": 1}}], lowWatermark=null, highWatermark=null) fail.
at org.apache.seatunnel.connectors.cdc.base.source.reader.external.IncrementalSourceScanFetcher.checkReadException(IncrementalSourceScanFetcher.java:216)
at org.apache.seatunnel.connectors.cdc.base.source.reader.external.IncrementalSourceScanFetcher.pollSplitRecords(IncrementalSourceScanFetcher.java:117)
at org.apache.seatunnel.connectors.cdc.base.source.reader.IncrementalSourceSplitReader.fetch(IncrementalSourceSplitReader.java:75)
... 8 more
Caused by: org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.exception.MongodbConnectorException: ErrorCode:[COMMON-06], ErrorDescription:[Illegal argument] - Execute snapshot read subtask for mongodb split SnapshotSplit(tableId=playground.tasks, splitKeyType=ROW<_id INT>, splitStart=[{"_id": 1}, {"_id": {"$minKey": 1}}], splitEnd=[{"_id": 1}, {"_id": {"$maxKey": 1}}], lowWatermark=null, highWatermark=null) fail
at org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.source.fetch.MongodbScanFetchTask.execute(MongodbScanFetchTask.java:156)
at org.apache.seatunnel.connectors.cdc.base.source.reader.external.IncrementalSourceScanFetcher.lambda$submitTask$0(IncrementalSourceScanFetcher.java:96)
... 5 more

at org.apache.seatunnel.core.starter.seatunnel.command.ClientExecuteCommand.execute(ClientExecuteCommand.java:220)
... 2 more
```

### Zeta or Flink or Spark Version

_No response_

### Java or Scala Version

_No response_

### Screenshots

### Are you willing to submit PR?

- [ ] Yes I am willing to submit a PR!

### Code of Conduct

- [x] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct)

Reproduce the MongoDB CDC job with the supplied Docker command and configuration, then inspect connectors/seatunnel-cdc/mongodb/.../MongodbScanFetchTask.java at the reported line 156 and the IncrementalSourceReader stack. Verify that INITIAL completes the snapshot and that TIMESTAMP consumes CDC events with nonzero read and write statistics.

Written by the indexing model from the issue text.