DanielLeens commented on PR #11206: URL: https://github.com/apache/seatunnel/pull/11206#issuecomment-6538373084
Pushed `d4faf1ea1aab8414aea81371c7c5112078f96739` (merge of `dev` `3ac2d1cec6a`, no conflicts, plus one fix commit). Compile/tests are left to CI per the validation policy; the fork Build for this head is [run 41164176211](https://github.com/DanielLeens/seatunnel/actions/runs/41164176211) (queued at posting time). No `pom.xml`, workflow or unrelated-connector changes. ### Unit test root cause (`MultiTableSinkWriterTest.testRuntimeRegistrationFailurePolicy:234`) The thrown `[MULTI_TABLE_ISOLATED_FAILURE] ... table=db1.unavailable, phase=checkpoint` comes from `MultiTableSinkWriter.currentSubSinkFailure` (`seatunnel-api/.../MultiTableSinkWriter.java:466-470`, dev code from #10600): under `CONTINUE_OTHER_TABLES`, when `sinkPrimaryKeys` is empty and `failedTables` is not, the coordinator reports the all-tables-isolated state as an isolated-failure `IOException` (the same contract `testRuntimeWriteZeroRetryKeepsImmediateIsolation` asserts). The test built the sink with an empty startup writer map, so after the only runtime table was quarantined there was no live table left and the rule fired inside `applySchemaChange`. The production policy path is correct and unchanged; the test now starts with one healthy table (`db1.healthy`), keeps every previous assertion (3 attempts, quarantine checkpointed, no runtime table recorded) and additionally asserts that the healthy table keeps writing and no row of the rejected table reaches a writer. This an swers @SEZ9's question: the branch is the non-fail-fast policy path, not F2 (unsupported sink) or F5 (retry/idempotency). ### Issue 1 (P0, two-replica deadlock) — `MultiTableSinkWriter.registerNewlyCreatedTable`, `MultiTableWriterRunnable.registerWriterAtSchemaChangeBarrier` Design note: registration runs only on the barrier dispatcher, i.e. the last queue worker inside `SchemaChangeBarrier.reachBarrier()`. At that point every worker, the dispatcher included, is parked in `MultiTableWriterRunnable.run()` holding its own runnable monitor, and every coordinator path that mutates the per-queue registries (`snapshotState`, `prepareCommit`, `aggregatedFlush`, `close`, `removeTableWriters`) blocks on those monitors until the latch is released. The barrier therefore already gives the dispatcher exclusive access; the two `synchronized (runnable.get(i))` blocks and the synchronized `registerWriter` were the only way to wait on a parked worker, and that is the lock/latch cycle. Both were removed: the registries are installed without any runnable monitor and the runnable exposes a package-private, unsynchronized `registerWriterAtSchemaChangeBarrier` whose Javadoc states the barrier-only contract and the happens-before provided by the latch count-down. No other s ynchronization was touched. Regression: `testRuntimeCreateTableRegistersOnEveryQueueWithoutDeadlock` (two queues, one CREATE, rows hashed to both queues, checkpoint with two writer states, bounded by `assertTimeoutPreemptively(30s)`; the previous code hangs it). ### Issue 2 (P0, runtime table lost on automatic recovery) — `SeaTunnelRowDebeziumDeserializeSchema.restoreCheckpointProducedType/emitPendingRestoreSchemaEvents`, `IncrementalSourceReader.admitCheckpointedRuntimeTables` Design note: with `scan.binlog.newly-added-table.enabled`, a checkpoint table absent from the submitted catalog is no longer treated as deleted: its checkpointed schema is admitted into `tables`/`tableRowConverters` and a `CreateTableEvent` is emitted before the first resumed record, so the engine row-type map and the sink writer are re-established (sinks treat the replayed CREATE as idempotent; JDBC is create-if-missing). While verifying the review's recovery path I found one more gap on the same path: the restored split's `tableIds` never contained the runtime table, and `JdbcSourceFetchTaskContext.registerDatabaseHistory` registers Debezium history only for `tableIds`, so after recovery the resumed rows of that table would hit Debezium's "schema isn't known" handling (`inconsistent.schema.handling.mode`, FAIL by default) before reaching the deserializer. The reader's existing restore reconciliation (`addSplits` -> prune against live captured tables) now also appends tables that exist only in the checkpointed produced type (not in the submitted catalog, not in `tableIds`) to the split's captured ids, after pruning, so a table dropped or excluded while the job was stopped stays out. For jobs without runtime admission the checkpointed tables are always a subset of the submitted catalog, so that step is a no-op. Opt-out is preserved: restoring with the option off keeps the legacy "deleted" path. Regressions: `testRestoreAdmitsCheckpointedRuntimeTableAbsentFromSubmittedCatalog` (deserializer initialized with A only, checkpoint A+B, B DML -> CREATE(B) then row), `testRestoreKeepsDroppingUnknownCheckpointTableWhenRuntimeDiscoveryDisabled`, and `IncrementalSourceReaderTest.testAddSplitsAdmitsCheckpointedRuntimeTableIntoCapturedTableIds`. An engine-level master-failover E2E is not added in this push; the Zeta restore path above is covered at the unit level and left for CI/E2E follow-up. ### Issue 3 (High, include list suppresses CREATE after admission) — `SeaTunnelRowDebeziumDeserializeSchema.emitPendingCreateTableEvents`, `SchemaChangeEventFilter` Design note (option chosen: required control event, documented filter contract): the CREATE of a runtime-admitted table is what creates the downstream row type and sink writer that the table's following rows depend on, so it is not a filterable schema change. `emitPendingCreateTableEvents` no longer applies the include/exclude filter, and `SchemaChangeEventFilter.filter` passes `CreateTableEvent` through unconditionally (class Javadoc updated). Admission and emission are now consistent at the source: an admitted table always gets its CREATE before DML; a rejected table (unsafe identifier/type) gets neither. The en/zh docs state that the lists only govern column-level DDL of captured tables. Regression: `testIncludeFilterDoesNotSuppressRuntimeCreateTableBeforeDml` (`SchemaChangeEventFilter.fromConfig` with `include=["add.column"]`, CREATE record then DML -> `create:db1.new_table`, `row:db1.new_table` in that order). ### SEZ9 open points (10-30 status comment) - F1/F4/F8 (identifier handling, target scope): in-code enforcement is `MySqlCatalogTableUtils.toRuntimeCatalogTable` (`connector-cdc-mysql/.../utils/MySqlCatalogTableUtils.java:47-48,124-128`): runtime identifiers must match `[A-Za-z0-9_$]*[A-Za-z_$][A-Za-z0-9_$]*` (digit-leading names such as `2024_orders` are accepted, all-numeric/backtick/semicolon names rejected with a WARN, no CREATE and no rows emitted); unsupported MySQL types are rejected there too (`:98-104`). Target scope: `MultiTableSinkWriter.hasSourceMatchedWriter` rejects a CREATE whose identity does not match its catalog table before any factory call (`testRuntimeCreateRejectsMismatchedIdentityBeforeFactory`), CREATE is dispatched only to the source-matched writer (`collectSchemaChangeDispatchTargets`, never to shared-target siblings), and the JDBC target comes only from the sink's configured `database`/`table` template, never from the source CREATE (`JdbcSinkFactory`, `AbstractJdbcSinkWriter` resolved-target save mode). Nothing is pending there. - F3 (`scan.newly-added-table.enabled` default `true`): kept `true` deliberately; the published `v3.0.0` `IncrementalSource.restore` already appended newly matched tables on hybrid restore, so `true` is the no-behaviour-change default for existing wildcard jobs and explicit `false` is the new opt-out (`MySqlIncrementalSource.isScanNewlyAddedTableEnabledOnRestore`, docs option table and "Newly Added Tables"). Runtime binlog discovery is a separate option that defaults to `false`. - F6 (`MultiTableState` format): `serialVersionUID` is pinned to the pre-change value (`5992121739651030596L`), `runtimeCreatedTables` is a new trailing field with a null-safe getter, so an old checkpoint restores with an empty runtime-table list (asserted by the legacy-state test). Rollback: a checkpoint written by this version that contains runtime tables is not readable by a version without the field semantics; the docs instruct keeping a pre-upgrade checkpoint and planning binlog retention before downgrading. - F7: `docs/en/connectors/sink/Jdbc.md` "Runtime tables from MySQL CDC" (`:57-59`) already states `is_exactly_once = false` and `generate_sink_sql = true` without a custom `query`; `JdbcSinkFactory` enforces it at `:469-477`. Tests left to CI: `MultiTableSinkWriterTest`, `SeaTunnelRowDebeziumDeserializeSchemaTest`, `IncrementalSourceReaderTest` plus the engine-v2-it and mysql-cdc-connector-it lanes that were red on the previous head must be re-evaluated on this exact head. Self-authored PR note: this comment is from the PR author account and carries no approval weight. -- 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]
