davidzollo opened a new issue, #12983: URL: https://github.com/apache/seatunnel/issues/12983
## Background Related: #11932. The broader Flink checkpointed schema-evolution protocol is tracked separately in #11958. PR #11932 restores the CDC deserializer and Zeta source collector from checkpoint schema state before restored records are emitted. Its documented runtime guarantee is limited to Zeta; the Flink/Spark translation collectors still inherit the default no-op `Collector.restoreSchema(List<CatalogTable>)`. This follow-up records the adapter work requested in the PR review. It does not claim that either adapter has been implemented or runtime-validated by #11932. ## Source evidence At PR head `f123d28c0dffb4ade2044a43ca1af058c28d332a`: - `seatunnel-api/src/main/java/org/apache/seatunnel/api/source/Collector.java`: `restoreSchema(List<CatalogTable>)` defaults to a no-op and documents the Zeta-only implementation. - `seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/reader/IncrementalSourceReader.java`: recovered collector schemas are selected only for schema-change-enabled jobs with a schema-aware deserializer, then supplied before polling restored records. - `seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/main/java/org/apache/seatunnel/translation/flink/source/FlinkRowCollector.java`: `collect(SeaTunnelRow)` forwards rows to ReaderOutput; `collect(SchemaChangeEvent)` emits an event row. The class has no `restoreSchema` override. - `seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-common/src/main/java/org/apache/seatunnel/translation/spark/serialization/InternalRowCollector.java`: `collect(SeaTunnelRow)` converts through its existing `InternalRowConverter` and hands off the InternalRow. The class has no `restoreSchema` override. The inherited no-op alone does not establish a runtime defect in every Flink/Spark path: the adapters own different type/conversion state. That ownership and the required recovery behavior must be demonstrated with real adapter tests. ## Bounded scope and acceptance criteria 1. For each adapter, trace where recovered source and downstream conversion schemas are owned. Define how `Collector.restoreSchema` participates; implement the required local-state refresh, or demonstrate with tests why no collector-local refresh is necessary. 2. Exercise a real CDC job that evolves a schema, completes a checkpoint/savepoint, restores, and processes the first post-restore record. Verify exact restored columns, types and values through the adapter and sink; include multi-table routing when supported. 3. Preserve the live-schema behavior of jobs with `schema-changes.enabled = false`, old checkpoints without table-level schema metadata, and each engine's checkpoint/commit boundaries. Do not replay duplicate downstream DDL merely to update collector-local state. 4. Cover recovery/failure boundaries and run the relevant Flink/Spark integration suites on the final change head. Keep Zeta validation and Flink/Spark validation reported separately. No adapter production change is included in this tracking issue or claimed completed here. -- 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]
