wuhainan commented on issue #12348: URL: https://github.com/apache/seatunnel/issues/12348#issuecomment-6210579647
> 3\. Thank you for the complete restore trace. **Classification: A / Connector-V2 MySQL-CDC recovery state migration.** > This is not just a missing initial snapshot. A configuration-added table must be introduced into the restored schema, snapshot, watermark, and incremental-split state as one compatible migration; otherwise the already restored reader can receive a binlog event for a table whose Debezium schema was never initialized. > I checked the current recovery paths and the related work. `BaseChangeStreamTableSourceFactory` restores checkpoint table structures, while `IncrementalSplitAssigner.addSplits()` only restores watermarks for table IDs present in the recovered incremental split. The existing [#11206](https://github.com/apache/seatunnel/pull/11206) work addresses runtime wildcard discovery in a running job; it does not prove the distinct A -> checkpoint/savepoint -> add B to configuration -> restore contract described here, and its Build is not green. I found no open PR that covers this root cause. > Before any implementation, please make that exact sequence a deterministic MySQL CDC E2E regression: restore A from its saved offset; verify B gets one snapshot at a defined watermark; then verify B binlog updates are accepted while A neither re-snapshots nor loses/replays its restored stream. The test must run for both checkpoint and savepoint recovery. A patch that only adds B to an `IncrementalSplit` is not sufficient unless it also establishes B schema/history and the snapshot-consistent watermark before incremental consumption. No help-wanted or assignment decision is made in this round. @DanielLeens I reproduced the same failure with SeaTunnel **3.0.0** using a minimal two-table test on 2026-10-16. The job fails immediately after adding table B to the configuration and restoring from the savepoint; no post-restore writes are needed to trigger it. ### Environment - SeaTunnel: **3.0.0** (`seatunnel-flink-20-starter.jar`; connector stack traces also report 3.0.0). - Flink: **1.20.0-h0.cbu.mrs.360.r18** (Huawei MRS distribution). - Java: **17.0.18**. - Deployment: **Flink YARN Application**, submitted through an internal platform. - Source: MySQL CDC; target: TiDB through JDBC / MultiTableSink. - Parallelism: **1**. - `startup.mode = "initial"`. - `schema-changes.enabled = true`. - Configured checkpoint interval/timeout: **30000 / 600000 ms**. - The parsed configuration contains `checkpoint.mode = "EXACTLY_ONCE"`, but the CDC fetcher logs **`exactly-once: false`**. I am reporting both values rather than assuming connector exactly-once was active. - The effective Flink restart strategy is **3 fixed-delay retries, 10000 ms apart**, despite failure-rate settings appearing in the parsed SeaTunnel config. This is a manual **savepoint recovery** reproduction. Checkpoint recovery and an automated E2E regression have not been verified in this test. ### Minimal reproduction 1. Create both source tables before the first job starts. B exists and contains historical rows, but is initially excluded from the capture list: ```sql CREATE DATABASE st_restore_src; CREATE TABLE st_restore_src.table_a ( id BIGINT NOT NULL PRIMARY KEY, val VARCHAR(100) NOT NULL ); CREATE TABLE st_restore_src.table_b ( id BIGINT NOT NULL PRIMARY KEY, val VARCHAR(100) NOT NULL ); INSERT INTO st_restore_src.table_a VALUES (1, 'a_initial_1'), (2, 'a_initial_2'); INSERT INTO st_restore_src.table_b VALUES (1, 'b_history_1'), (2, 'b_history_2'); ``` 2. Prepare the target database/tables: ```sql CREATE DATABASE st_restore_dst; CREATE TABLE st_restore_dst.table_a ( id BIGINT NOT NULL PRIMARY KEY, val VARCHAR(100) NOT NULL ); CREATE TABLE st_restore_dst.table_b ( id BIGINT NOT NULL PRIMARY KEY, val VARCHAR(100) NOT NULL ); ``` 3. Start a streaming MySQL-CDC job with only A: ```hocon database-names = ["st_restore_src"] table-names = ["st_restore_src.table_a"] table-names-config = [ { table = "st_restore_src.table_a", primaryKeys = ["id"] } ] startup.mode = "initial" schema-changes.enabled = true ``` The JDBC sink uses generated upsert SQL, target database `st_restore_dst`, and `table = "${table_name}"`. 4. Wait for A's initial data to arrive. Update/insert A, verify incremental synchronization, and wait for a successful checkpoint: ```sql UPDATE st_restore_src.table_a SET val = 'a_before_savepoint' WHERE id = 1; INSERT INTO st_restore_src.table_a VALUES (3, 'a_before_savepoint_3'); ``` 5. Stop the job with a savepoint through the platform and retain its path. This run subsequently restores **Savepoint 10**. 6. While the job is stopped, write changes to A and B: ```sql UPDATE st_restore_src.table_a SET val = 'a_changed_while_stopped' WHERE id = 1; INSERT INTO st_restore_src.table_a VALUES (4, 'a_inserted_while_stopped'); DELETE FROM st_restore_src.table_a WHERE id = 2; INSERT INTO st_restore_src.table_b VALUES (3, 'b_history_while_stopped'); ``` 7. Add B to the same source configuration and restore from the savepoint, retaining parallelism 1 and the same runtime: ```hocon database-names = ["st_restore_src"] table-names = [ "st_restore_src.table_a", "st_restore_src.table_b" ] table-names-config = [ { table = "st_restore_src.table_a", primaryKeys = ["id"] }, { table = "st_restore_src.table_b", primaryKeys = ["id"] } ] startup.mode = "initial" schema-changes.enabled = true ``` **The job fails at step 7**, before the post-restore DML checks can be performed. B already has a binlog event from step 6, so restoring the old stream is sufficient to trigger the failure. ### Key logs The following excerpts are from the same run (timestamps UTC+8). Internal paths/addresses are redacted; the long incremental-split line is shortened. **1. Flink restores the savepoint:** ```text 2026-10-16 15:13:39,458 INFO CheckpointCoordinator - Restoring job ffffffffa8039e850000000000000000 from Savepoint 10 @ 0 for ffffffffa8039e850000000000000000 located at <savepoint-path>. ``` **2. The enumerator discovers B as a pending snapshot table:** ```text 2026-10-16 15:13:39,602 INFO SnapshotSplitAssigner - SnapshotSplitAssigner created with remaining tables: [st_restore_src.table_b] 2026-10-16 15:13:39,603 INFO SnapshotSplitAssigner - SnapshotSplitAssigner created with remaining splits: [] 2026-10-16 15:13:39,603 INFO SnapshotSplitAssigner - SnapshotSplitAssigner created with assigned splits: [] ``` **3. The reader restores only A's incremental state and history:** ```text 2026-10-16 15:13:51,335 INFO IncrementalSourceReader - subtask 0 add splits: incremental-split-0 2026-10-16 15:13:51,481 INFO IncrementalSourceReader - The incremental split[incremental-split-0] has 1 checkpoint table(s) for restore: [st_restore_src.table_a]. 2026-10-16 15:13:51,482 INFO IncrementalSourceReader - The incremental split[incremental-split-0] has checkpoint history table changes for restore. 2026-10-16 15:13:51,656 INFO IncrementalSourceStreamFetcher - Start incremental read task for incremental split: IncrementalSplit( tableIds=[st_restore_src.table_a], startupOffset={transaction_id=null, ts_sec=1792134501, file=binlog.000459, pos=665187302, server_id=1}, ... completedSnapshotSplitInfos=[], checkpointDataType=null, checkpointTables=[...st_restore_src.table_a...], historyTableChanges={st_restore_src.table_a=...} ) exactly-once: false ``` There is no B snapshot-split assignment/execution in the supplied TaskManager log before the failure. **4. B's TABLE_MAP event causes the source to fail:** ```text 2026-10-16 15:13:51,755 ERROR MySqlStreamingChangeEventSource - Encountered change event ... eventType=TABLE_MAP ... database='st_restore_src', table='table_b' ... for table st_restore_src.table_b whose schema isn't known to this connector. 2026-10-16 15:13:51,755 ERROR MySqlStreamingChangeEventSource - Error during binlog processing. Last offset stored = {transaction_id=null, ts_sec=1792134691, file=binlog.000459, pos=733312178, server_id=1, event=1}, binlog reader near position = binlog.000459/733312249 Caused by: io.debezium.DebeziumException: Encountered change event for table st_restore_src.table_b whose schema isn't known to this connector ``` **5. Retries hit the same event; the job ultimately fails:** ```text 2026-10-16 15:14:30,443 INFO ExecutionGraph - Job mysql2tidb_multi_table_restart_from_savepoint_test (ffffffffa8039e850000000000000000) switched from state RUNNING to FAILING. org.apache.flink.runtime.JobException: Recovery is suppressed by FixedDelayRestartBackoffTimeStrategy( maxNumberRestartAttempts=3, backoffTimeMS=10000) 2026-10-16 15:14:30,455 INFO ExecutionGraph - Job mysql2tidb_multi_table_restart_from_savepoint_test (ffffffffa8039e850000000000000000) switched from state FAILING to FAILED. ``` ### Expected vs. observed Expected: A continues from its saved offset, including changes made while stopped; B receives its initial snapshot and then incremental changes. Observed: the enumerator recognizes B, but the restored reader starts the old A-only incremental split, and processing B's binlog event fails because its schema is missing. This reproduces the recovery-state inconsistency described in this issue on 3.0.0. The supplied logs establish the source failure. I have not included target row-count/data-loss assertions or claims of duplicate-free recovery, because those have not been independently verified for this failed run. -- 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]
