sd4324530 opened a new pull request, #4560: URL: https://github.com/apache/flink-cdc/pull/4560
## What is the purpose of this pull request? Fix [FLINK-40795](https://issues.apache.org/jira/browse/FLINK-40795): when a table is removed from a pipeline's capture list and later added back while its upstream schema drifted in between (e.g. a column was dropped while out of capture), the downstream table is never re-created and writes fail with metadata mismatch errors. The source side works as intended: the enumerator purges the removed table from its state, so the re-added table is detected as newly-added, re-snapshotted, and a `CreateTableEvent` carrying the latest schema is emitted. The problem is in the regular `SchemaCoordinator`: - `SchemaManager` state (checkpointed) never removes schemas of tables that left the pipeline, so a stale schema of the removed table survives across savepoints; - `SchemaUtils#isSchemaChangeEventRedundant` decides "redundant" for a `CreateTableEvent` based only on the *presence* of a recorded schema, never comparing it with the incoming one, so the drifted `CreateTableEvent` is silently dropped: the `MetadataApplier` is never invoked (table not re-created), and the stale evolved schema keeps being pushed back to `SchemaOperator`s and sink writers (records coerced into the stale shape -> metadata mismatch at the sink). The redundancy check was introduced by FLINK-36690 to skip duplicated `CreateTableEvent`s in the snapshot stage; skipping *identical* duplicates remains correct and is preserved. ## Brief change log - `SchemaUtils#isSchemaChangeEventRedundant`: a `CreateTableEvent` is now redundant only when the recorded schema is **identical** to the incoming event's schema. - `SchemaCoordinator#deduceEvolvedSchemaChanges` (regular topology): when a `CreateTableEvent` targets a sink table that already has an evolved schema (re-snapshot after removal, or an ALTER landing mid-snapshot), emit `CreateTableEvent(targetSchema)` (idempotent create-if-absent, covering an externally dropped downstream table) followed by alignment events derived via `SchemaMergingUtils#getSchemaDifference`, instead of forwarding the raw event. The target schema honors the configured behavior: - `EVOLVE` / `TRY_EVOLVE` / `EXCEPTION`: the incoming schema; - `LENIENT`: `getLeastCommonSchema` (never narrows the downstream schema); - `IGNORE`: the frozen current schema. - Only the resulting `CreateTableEvent` is propagated downstream (it carries the full target schema); alignment events are applied to the external system only, and evolved-schema state updates are guarded by the same redundancy check so alignment events cannot corrupt the state by being applied twice. ## Verifying this change This change added tests and can be verified as follows: - Added coordinator-level tests in `SchemaCoordinatorTest` (regular): - `reAddedTableWithDroppedColumnRecreatesDownstreamTable` (EVOLVE: re-creates with the incoming schema and aligns via `DropColumnEvent`); - `reAddedTableWithAddedColumnAlignsDownstreamTable` (EVOLVE: aligns via `AddColumnEvent`); - `lenientBehaviorKeepsDroppedColumnOnReAddedTable` (LENIENT: downstream schema never narrows); - `ignoreBehaviorFreezesEvolvedSchemaOnReAddedTable` (IGNORE: schema frozen, table still re-created if absent); - `duplicateCreateTableEventWithIdenticalSchemaIsSkipped` (regression guard for the FLINK-36690 dedup semantics). - Added `SchemaUtilsTest#testIsSchemaChangeEventRedundantOnCreateTableEvent` for the utility change. - All new tests fail on master without the fix (the applier receives no events) and pass with it. - Full suites green: `flink-cdc-runtime` (988 tests) and `flink-cdc-common` (96 tests) on Flink 1.20; schema operator tests (58) green under `-Pflink2`; `spotless:check` and `checkstyle:check` pass. - Connector call sites of the changed utility verified: `PaimonMetadataApplierTest` (25), `FlussEventSerializationSchemaTest` (2), `HudiMetadataApplierTest` (3) green. Distributed-topology tests are unaffected (17 green). ## Documentation - Does this pull request introduce a new feature? (yes / no) **no** - If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented) **not applicable** --- ##### Was generative AI tooling used to co-author this PR? - [X] Yes (Claude Code) Generated-by: Claude Code Qwen3.8-max -- 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]
