nielifeng commented on issue #10461: URL: https://github.com/apache/seatunnel/issues/10461#issuecomment-5020733613
I reproduced this issue on Zeta with a MySQL CDC streaming job and real Worker failures. Environment: - SeaTunnel `3.0.0-SNAPSHOT`, commit `46d2f3909f7e97f8f516c8fba68d12df471d7718` - MySQL 8.0.46 - 3 Masters, 2 Workers, fixed slots - `MySQL-CDC -> Console` and `MySQL-CDC -> JDBC MySQL` were both tested Reproduction sequence: 1. Start from a four-column table: `id, name, score, updated_at`. 2. Start the CDC job and wait for normal checkpoints. 3. Execute `ADD COLUMN email VARCHAR(128) AFTER name`. 4. Confirm both schema-change-before and schema-change-after checkpoints complete. 5. Kill the Worker running the source task. 6. Insert a row containing the new `email` column after recovery. With the original code, the retry split contained five-column `checkpointTables` and nonempty `historyTableChanges`, but the reader skipped restoring them because `checkpointDataType` was null. The restored reader continued with the old four-column schema. After another retry, Debezium failed with conversion errors such as `Failed to properly convert data value`, followed by `Error during binlog processing` and `Producer failure`. There is also a second layer to the bug: restoring only the Debezium deserializer/table history is not sufficient. In that version, the first five-column row after recovery failed with: ```text java.lang.ArrayIndexOutOfBoundsException: Index 4 out of bounds for length 4 at SeaTunnelRowType.getFieldType at SeaTunnelRow.getBytesSize at SeaTunnelSourceCollector.collect ``` The deserializer had correctly restored five columns, but `SeaTunnelSourceCollector` and the sink writer were still initialized from the original four-column DAG schema. The complete fix I validated does the following: 1. Restore nonempty `checkpointTables` and `historyTableChanges` independently of the deprecated `checkpointDataType` field. 2. Replace the stale Debezium history map with checkpoint history. 3. Before the first restored data record, emit a dedicated runtime schema-restore event carrying the full checkpoint `CatalogTable`. 4. Update source/sink runtime row types from that event. 5. For JDBC sinks, rebuild the writer using the restored schema without applying physical DDL again. Real validation results: - Console path: two consecutive Worker failures and recoveries. - Both post-recovery rows retained their `email` values. - Final checkpoint counts before manual stop: triggered 139, completed 139, failed 0, restored 2. - JDBC MySQL path: target `ADD COLUMN` executed exactly once before failure. - On recovery JDBC logged that it restored runtime schema without applying physical DDL; no duplicate-column error occurred. - JDBC final checkpoint counts before manual stop: triggered 66, completed 66, failed 0, restored 1. - No `ArrayIndexOutOfBoundsException`, Debezium conversion error, `Producer failure`, or binlog processing error in either fixed run. - Full `connector-cdc-base` test suite: 66 tests passed. I have a source patch and can prepare a PR with the focused tests and the two real-test scenarios above. -- 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]
