wuhainan opened a new issue, #11260: URL: https://github.com/apache/seatunnel/issues/11260
### 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 When restoring a MySQL-CDC multi-table streaming job from a savepoint/checkpoint after removing one table from the source table list, the job fails during source reader startup. The removed table is no longer included in the current SeaTunnel config, but the restored CDC source state still contains this table in the incremental split state. When the CDC source reader starts, it restores Debezium schema history by iterating over `IncrementalSplit.getTableIds()`. The error is: `Can't obtain schema for table alpha_online.account_interests` Expected behavior: When restoring a multi-table CDC job with fewer tables than the previous checkpoint/savepoint, SeaTunnel should prune removed tables from all restored source states, including incremental split state, so the remaining tables can continue from the previous offset. Actual behavior: `IncrementalSource.restore()` removes deleted tables from `SnapshotPhaseState`, including `remainingTables`, `alreadyProcessedTables`, `remainingSplits`, `assignedSplits`, and `splitCompletedOffsets`. However, the restored `IncrementalSplit.tableIds` still contains the removed table. The reader then tries to query schema for that removed table and fails. Related code: - `IncrementalSource.restore()` prunes deleted tables only from `SnapshotPhaseState`: https://github.com/apache/seatunnel/blob/dev/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/IncrementalSource.java#L395-L451 - `JdbcSourceFetchTaskContext.registerDatabaseHistory()` iterates over restored `IncrementalSplit.getTableIds()` and queries schema: https://github.com/apache/seatunnel/blob/dev/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/reader/external/JdbcSourceFetchTaskContext.java Possible fix: When restoring CDC state, deleted tables should also be pruned from incremental split related states: - `IncrementalSplit.tableIds` - `completedSnapshotSplitInfos` - `checkpointTables` - `historyTableChanges` If all table IDs in a restored incremental split are removed, the split should be dropped. ### SeaTunnel Version 2.3.13 ### SeaTunnel Config ```conf # Before: the job contained two tables. # # source.table-names: # - alpha_online.account_histories # - alpha_online.account_interests # # source.table-names-config: # - table = alpha_online.account_histories, primaryKeys = ["id"] # - table = alpha_online.account_interests, primaryKeys = ["id"] # # transform.FieldMapper.table_transform: # - alpha_online.account_histories # - alpha_online.account_interests # # After: alpha_online.account_interests was removed from source table list and transform mapping, # then the job was restored from the previous savepoint/checkpoint. env = { parallelism = 1 job.mode = "STREAMING" job.name = "alpha_online_multi_table_sync_test" checkpoint.interval = 30000 checkpoint.timeout = 600000 checkpoint.mode = "EXACTLY_ONCE" restart-strategy = "failure-rate" restart-strategy.failure-rate.max-failures-per-interval = 10 restart-strategy.failure-rate.failure-rate-interval = "300 s" restart-strategy.failure-rate.delay = "10 s" jobmanager.memory.process.size = "1024m" taskmanager.memory.process.size = "2048m" taskmanager.numberOfTaskSlots = 1 allowNonRestoredState = "true" } source = { MySQL-CDC = { hostname = "******" port = 3306 url = "jdbc:mysql://******:3306/" username = "******" password = "******" database-names = ["alpha_online"] table-names = [ "alpha_online.account_histories" ] table-names-config = [ { table = "alpha_online.account_histories" primaryKeys = ["id"] } ] server-id = 5808 startup.mode = "initial" server-time-zone = "Asia/Shanghai" plugin_output = "src" } } ``` ### Running Command ```shell # Submitted by platform through Flink YARN application mode. ``` ### Error Exception ```log 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.translation.flink.source.FlinkSourceReader.pollNext(FlinkSourceReader.java:99) at org.apache.flink.streaming.api.operators.SourceOperator.emitNext(SourceOperator.java:422) at org.apache.flink.streaming.runtime.io.StreamTaskSourceInput.emitNext(StreamTaskSourceInput.java:68) at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65) at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:638) at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:231) at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:973) at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:917) at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:970) at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:949) at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:763) at org.apache.flink.runtime.taskmanager.Task.run(Task.java:575) at java.lang.Thread.run(Thread.java:840) 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) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539) at java.util.concurrent.FutureTask.run(FutureTask.java:264) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) Caused by: java.lang.RuntimeException: Can't obtain schema for table alpha_online.account_interests at org.apache.seatunnel.connectors.seatunnel.cdc.mysql.utils.MySqlSchema.readTableSchema(MySqlSchema.java:102) at org.apache.seatunnel.connectors.seatunnel.cdc.mysql.utils.MySqlSchema.getTableSchema(MySqlSchema.java:76) at org.apache.seatunnel.connectors.seatunnel.cdc.mysql.source.MySqlDialect.queryTableSchema(MySqlDialect.java:110) at org.apache.seatunnel.connectors.cdc.base.source.reader.external.JdbcSourceFetchTaskContext.registerDatabaseHistory(JdbcSourceFetchTaskContext.java:200) at org.apache.seatunnel.connectors.seatunnel.cdc.mysql.source.reader.fetch.MySqlSourceFetchTaskContext.configure(MySqlSourceFetchTaskContext.java:119) at org.apache.seatunnel.connectors.cdc.base.source.reader.external.IncrementalSourceStreamFetcher.submitTask(IncrementalSourceStreamFetcher.java:98) at org.apache.seatunnel.connectors.cdc.base.source.reader.IncrementalSourceSplitReader.checkSplitOrStartNext(IncrementalSourceSplitReader.java:186) at org.apache.seatunnel.connectors.cdc.base.source.reader.IncrementalSourceSplitReader.fetch(IncrementalSourceSplitReader.java:84) at org.apache.seatunnel.connectors.seatunnel.common.source.reader.fetcher.FetchTask.run(FetchTask.java:54) at org.apache.seatunnel.connectors.seatunnel.common.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:162) at org.apache.seatunnel.connectors.seatunnel.common.source.reader.fetcher.SplitFetcher.run(SplitFetcher.java:81) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539) at java.util.concurrent.FutureTask.run(FutureTask.java:264) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) ``` ### Zeta or Flink or Spark Version flink 1.20 ### Java or Scala Version jdk 17 ### Screenshots _No response_ ### Are you willing to submit PR? - [x] 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]
