uniquemj opened a new issue, #9950: URL: https://github.com/apache/seatunnel/issues/9950
### 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 <img width="1888" height="907" alt="Image" src="https://github.com/user-attachments/assets/002cf57a-ad5d-4d86-a275-4bb756395a4c"; /> ### 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) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
