ryanmeowy commented on issue #10203:
URL: https://github.com/apache/seatunnel/issues/10203#issuecomment-6528646084
@DanielLeens @SEZ9 — the revision with the three contract items I promised
on 24 Oct is below. It supersedes the earlier outline; the 23 Oct transition
table and the 2a/2b paragraph still stand and are not repeated here.
## Design revision — live discovery of new MySQL tables (#10203)
This is a design revision, published for review. The design is ready to be
agreed now; the implementation is deliberately queued behind #11206, for the
reasons in §8.3. Nothing here is an implementation PR.
**Baseline.** `dev` at `128994ca0`, which includes #13006 (`[Fix][CDC] Keep
per-table restored offsets when tables are added across a savepoint`, merged
2026-10-28). Every code reference below is against that baseline. Re-checked
against the current tip `3ac2d1cec`: the only CDC change on `dev` since the
baseline is `Db2IncrementalSourceFactory` and its test, and none of the files
cited here moved, so the line numbers still hold.
## 0. What changed since the previous note
1. #13006 is merged, so `IncrementalSplit` now carries `tableStartupOffsets`
and `IncrementalSourceStreamFetcher.configureFilter()` gives every table its
own start boundary (`IncrementalSplit.java:75`,
`IncrementalSplitState.java:50`,
`IncrementalSourceStreamFetcher.configureFilter:321-368`). The hand-off of
"where each already-read table resumes" no longer has to be invented by this
feature; it is the same mechanism.
2. #12567 is closed and superseded by #13006, so the earlier question about
taking that fix over is moot.
3. Row 3 must be re-worded: `maxSplitHighWatermarkMap` is now built from
`CompletedSnapshotSplitInfo` *and* per-table restored offsets
(`IncrementalSourceStreamFetcher.configureFilter:321-368`), and `shouldEmit()`
has an extra `position.isBefore(tableStartupOffset)` gate
(`IncrementalSourceStreamFetcher.shouldEmit:287-289`).
4. The completion signal is the split's own watermark pair, not a
finished-splits map (see contract 1).
5. Scope correction: this is now stated as a complement to #11206, not a
standalone phase 2. #11206 owns the runtime registration of a new table on the
deserializer and sink-writer side (`CreateTableEvent` does not exist on `dev`),
so the admission snapshot only helps while that registration is active; section
1 and section 5 say so.
6. Design and implementation are split on purpose: the three contracts and
the test plan can be agreed now, and the code is deliberately queued behind
#11206. §8.3 records why rather than leaving the hold unexplained, and §8.4
states the one thing this design asks of #11206's own recovery fix.
## 1. Scope
- What this adds: a table that starts matching `table_pattern` **while the
job is streaming** is discovered, backfilled once, and then read incrementally
— without restarting the job. (Startup-time discovery already exists;
restore-time discovery already exists; neither is touched.)
- **This feature complements #11206, it does not replace it.** #11206's
`scan.binlog.newly-added-table.enabled` registers a newly created table at
runtime and starts reading it at the `CREATE TABLE` binlog position, without
historical rows. What is missing — and what this proposes — is the backfill of
the rows that pre-date that position. So the reader hand-off in section 4 is
not the main risk; the main risk is that the rest of the pipeline has not
learned about `T` yet:
- The deserializer drops rows of a table it did not know at init:
`SeaTunnelRowDebeziumDeserializeSchema.deserializeDataChangeRecord:248-253`
looks up `tableRowConverters.get(tableId)` in multi-table mode and returns on a
miss (`"Ignore newly added table"`). `createTableRowConverters` (`:413-442`) is
rebuilt on a schema change (`:226-228`) and on restore (`:393-395`), but always
from the same `tables` list, which is fixed at construction (`:99`, `:103-105`)
and only mutated by those two paths — a table that appears at runtime never
enters it.
- The sink refuses to route them: `MultiTableSinkWriter.write:601-615`
throws `"multi table sink can not write table: "` for a `tableId` absent from
`sinkPrimaryKeys`, and only degrades to a per-table failure when
`failurePolicy.continueOtherTables()` is on. `sinkPrimaryKeys` is filled from
the writers built in `MultiTableSink.createWriter:143-156`, which iterates the
submission-time `sinks` map, so a table that did not exist at submission has no
writer either.
- `CreateTableEvent` / `CreateTableEventHandler` do not exist on `dev`;
they, the runtime `tableRowConverters` rebuild and the runtime sink-writer
registration are all introduced by #11206.
- The backfill client cannot register `T` on its own:
`MySqlSnapshotFetchTask.createBackfillBinlogReadTask:143-160` forces
`table.include.list = <T>` and `include.schema.changes = false`, so a `CREATE
TABLE` seen there is not parsed into a schema.
- Consequence for the design: enabling this mode requires
`scan.binlog.newly-added-table.enabled = true` so that #11206's registration
path is active. If it is off, the admission snapshot can complete and every row
of `T` is then discarded downstream. Section 5 rejects that combination rather
than letting it run silently.
- v1 connector scope: **MySQL only**. OceanBase inherits the MySQL option
rule but gets an explicit config-time rejection in v1 (section 5).
- Default off. Existing behaviour must be unchanged when the option is off.
## 2. State model and the one-checkpoint rule
The enumerator owns an admission registry, persisted as part of its
checkpoint state (fields on `IncrementalPhaseState`, null-safe for checkpoints
written before this feature):
- `pendingSnapshotTables` — admitted, snapshot split not finished yet.
- `admittedTables` — admitted and snapshot finished.
- `tableWatermarks[T] = W_high` — the high watermark captured for T's
admission snapshot.
- `tableStartupOffsets[T] = W_high` — the same value handed to the rebuilt
incremental split so that T keeps its floor after `completedSnapshotSplitInfos`
is cleared (section 4).
Rules:
- The enumerator is the only writer. A reader reports a finished split; it
never admits a table.
- `T` joins the incremental split's `tableIds` only in the checkpoint that
also carries `admittedTables[T]` and `tableWatermarks[T]`. Anything before a
completed checkpoint is lost on recovery and re-runs instead of resuming.
- A poller tick that sees `T` already in `pendingSnapshotTables` or
`admittedTables` is a no-op, so timer ticks, re-discovery and restore cannot
admit twice.
- The registry is enumerator state (`IncrementalPhaseState`, written by
`IncrementalSplitAssigner.snapshotState:241-243` and wrapped by
`HybridSplitAssigner.snapshotState:147-151`). The incremental split that
references T is reader state (`IncrementalSourceReader.snapshotState:459`).
Both are part of the same checkpoint barrier, so the split (with T's
`CompletedSnapshotSplitInfo`), T's `W_high` and the registry are restored
together; there is no checkpoint in which one side already references T and the
other does not.
### Completion signal (rows 2a/2b)
What marks a restored split complete is **its own watermark pair**, not
membership in a finished-splits map: `SnapshotSplit.isSnapshotReadFinished()`
(`SnapshotSplit.java:70-72`), carried by
`SnapshotSplitState.lowWatermark`/`highWatermark`
(`SnapshotSplitState.java:31-32`, written back by `toSourceSplit()` at
`:39-49`). `finishedUnackedSplits` is rebuilt from that pair on restore
(`IncrementalSourceReader.addSplits:152-153`), and
`SnapshotSplitAssigner.restoreCompletedSnapshotSplit` applies the same
predicate (`:317-329`).
### Mid-backfill cut
The HIGH watermark is dispatched before the bounded backfill, but it is not
checkpointed yet: `MySqlSnapshotSplitReadTask:154` writes it onto
`SnapshotSplitChangeEventSourceContext` (the backfill's stop offset), and both
state fields stay null until the emitter consumes the batch. On the
exactly-once path
`IncrementalSourceScanFetcher.pollSplitRecordsIfExactlyOnce:148-198` returns
`[low ... high]` only after the END watermark, so a checkpoint taken during the
backfill persists the split with both watermarks null and a restore re-runs it
as 2a. Once a checkpoint carries both watermarks, restore treats the split as
2b without replaying, whether or not `onSplitFinished` had run.
## 3. Contract 1 — atomic checkpoint persistence
- The registry entries and the reader's incremental split that references T
are persisted in the same checkpoint barrier. There is no intermediate state in
which the incremental split already contains `T` but the registry does not, or
the registry contains `W_high` but the split does not carry `T`'s
`CompletedSnapshotSplitInfo`.
- Invariants checked in unit tests: `pendingSnapshotTables ∩ admittedTables
= ∅`; `admittedTables ⊆ capturedTables`; `tableWatermarks.keySet() =
admittedTables`; a table's `W_high` is never below its low watermark.
- Admission protocol (enumerator): tick → discover matching tables not in
the registry → chunk T with its own splitter
(`dialect.createChunkSplitter(sourceConfig)`, as
`SnapshotSplitAssigner.open:150-151` does;
`chunkSplitter.generateSplits(nextTable)` at `:185`) → put `T` in
`pendingSnapshotTables` → dispatch the splits to the owning reader. The
low/high watermarks are captured by the scan task when the split runs, not by
the enumerator (`MySqlSnapshotSplitReadTask:137`, `:149`, `:154`); the reader's
completion report is what moves `T` from `pendingSnapshotTables` to
`admittedTables` and what supplies `W_high`. The incremental split is rebuilt
with T's `CompletedSnapshotSplitInfo` entry in a later checkpoint that carries
the registry. The snapshot-phase `SnapshotSplitAssigner` is already completed
by then, so the admission path cannot reuse its `getNext()`; it reuses the
chunk splitter and its own split bookkeeping.
- Restore: the subtraction basis becomes `capturedTables −
(checkpointCapturedTables ∪ admittedTables ∪ pendingSnapshotTables)`; the
re-drive path recreates a snapshot split from `pendingSnapshotTables` when no
reader holds it. This is an extension of `IncrementalSource.restore()`
(`:427-482`), which today only diffs `remainingTables` against the checkpointed
table set.
- Discovery cadence is an enumerator-owned timer, default 60 s, independent
of the checkpoint interval. There is no scheduling facility to build on:
`SourceSplitEnumerator.Context` (`SourceSplitEnumerator.java:99-152`) exposes
only `currentParallelism`, `registeredReaders`, `assignSplit`,
`signalNoMoreSplits`, `sendEventToSourceReader`, `getMetricsContext` and
`getEventListener` — no timer, no callback — and `IncrementalSourceEnumerator`
holds no executor; its `close()` (`:176-179`) only forwards to
`splitAssigner.close()`. So the enumerator creates its own single-thread
scheduler in `open()` and shuts it down in `close()`.
- Cost of a tick, and where it may run: one tick is one
`dialect.discoverDataCollections(sourceConfig)` round trip against
`information_schema` (the same call
`IncrementalSource.createEnumerator:314-315` makes once at startup, and
`SnapshotSplitAssigner.open:154-164` makes for the legacy state). It must not
run inside `synchronized (context)`: the engine takes that same monitor on the
job's barrier thread (`SourceSplitEnumeratorTask.triggerBarrier:146`, lock at
`:158-160`) and calls `snapshotState` there, so a slow metadata query would
stall checkpointing. The query runs outside the lock; only the registry
mutation and the split bookkeeping are done under it. The timer touches no
binlog and opens no second replication client, so the per-tick cost is one
metadata query per interval.
- Reproducibility of a re-driven admission split:
`chunkSplitter.generateSplits` samples when `split.allow-sampling` is on and
the shard count exceeds `sample-sharding.threshold`, with the rate from
`inverse-sampling.rate` (`AbstractJdbcSourceChunkSplitter:231-250`, options at
`JdbcSourceOptions.java:125`/`:136`/`:146`). The sample is every
`inverse-sampling.rate`-th row of an unordered scan
(`MySqlUtils.skipReadAndSortSampleData:142-192`), so the boundaries depend on
physical row order and on rows written since; a split rebuilt from
`pendingSnapshotTables` after a crash can therefore have different boundaries
and a different `splitId` than the one assigned before the crash. This is safe
only because the previous attempt read nothing that was checkpointed; if a
checkpoint stands between "assigned" and "finished", the split is recovered
from the reader's `splitStates` and *not* rebuilt. A rebuilt split deliberately
gets a fresh id for that reason.
- The DDL channel is not the correctness mechanism either: `CREATE TABLE`
events only reach a reader inside its binlog stream
(`IncrementalSourceStreamFetcher.SchemaChangeStreamSplitter:370`), so DDL
during a snapshot phase, a restart, or between admission and the next
checkpoint is invisible to it. Across a restore the timer restarts; the
registry diff on the next tick covers missed ticks.
## 4. Contract 2 — dispatch the admission snapshot only to the owner of the
live stream
- The enumerator tracks which reader owns the incremental split that carries
T's table, and assigns T's admission snapshot split only to that reader. With
the default `incremental.parallelism = 1`
(`SourceOptions.INCREMENTAL_PARALLELISM:75-79`) there is a single incremental
split, so the owner is simply the reader that holds the streaming split.
- Why it must be owner-only:
`IncrementalSourceSplitReader.checkSplitOrStartNext()` returns immediately
while `currentFetcher instanceof IncrementalSourceStreamFetcher` (`:152-156`),
and a stream fetcher never finishes, so a snapshot split handed to any other
reader would sit in its queue forever; `canAssignNextSplit()` is false for the
same reason (`:190`). Nor does a split arriving mid-stream start a scan on the
owner: `SourceReader.addSplits` → `SingleThreadFetcherManager.addSplits`
(`:53-63`) appends it to the running fetcher's queue. So the admission split
has to preempt the stream fetcher on the owner (close it, run the scan, re-open
the stream after the backfill), which is what the state machine below describes
and what the state-machine test in section 7 covers.
- The reader must also be re-armed, not just re-targeted. Once the
incremental split has been assigned the enumerator signals no-more-splits to
that reader (`IncrementalSourceEnumerator.assignSplits:204-215`), and
`SourceReaderBase.handleNoMoreSplits:140-142` latches `noMoreSplitsAssignment =
true` with no reset path — `addSplits` (`:126-137`) does not clear it. In a
*bounded* job (`stop.mode = latest`/`specific`,
`IncrementalSource.getBoundedness:254-258`) the reader can therefore observe
`isNoMoreSplitsAssignment() && isNoMoreElement()`
(`IncrementalSourceReader.pollNext:126-131`, `isNoMoreElement` at
`SourceReaderBase:150-153`) while the stream fetcher is closed and the queue is
empty, send `signalNoMoreElement` and finish the task before T's backfill runs.
The admission dispatch must therefore clear that latch (extend
`handleNoMoreSplits`/add a re-arm event), and the E2E must include a bounded
run. Default `stop.mode = never`
(`MySqlIncrementalSourceOptions.STOP_MODE:76-85`)
is unbounded and does not hit it.
- Reassignment: when the owning reader fails, Zeta calls `addSplitsBack`
(`RestoredSplitOperation` → `SourceSplitEnumeratorTask.addSplitsBack` →
enumerator); the enumerator re-binds ownership and re-dispatches both the
incremental split and the pending admission split to the new owner, re-arming
the new reader as above. The admission split must stay in the admission
registry's assigned-splits map for the whole read:
`restoreCompletedSnapshotSplit` only accepts a finished split that is *also*
still in `assignedSplits` (`SnapshotSplitAssigner:317-320`), so a split that
dropped out of the bookkeeping is re-enqueued and 2b silently degrades to 2a.
- Swapping the stream split: by the time T is admitted the incremental split
is already assigned (`IncrementalSplitAssigner.splitAssigned`, `:83`), so
`getNext()` returns empty (`:148-150`) and the assigner will not produce a
second stream split on its own. Admission therefore re-issues one split through
the same owner, never a second stream client: add T to
`context.getCapturedTables()` — that is the filter `addSplits` applies before
it records anything (`:193`, `:202`), so a table absent from it has its
`CompletedSnapshotSplitInfo` and `tableStartupOffsets` silently skipped — hand
a rebuilt split to the assigner's existing `addSplits` (`:183-238`, which
repopulates `tableWatermarks`/`tableStartupOffsets` and rebuilds
`splitCompletedOffsets` + `assignedSnapshotSplit` from each
`CompletedSnapshotSplitInfo`), and clear `splitAssigned` so `getNext()` builds
the new split — today that field is only written through a
`@VisibleForTesting`, package-private setter (`:389-392`), so
this needs a package-private admission entry point with a production caller
rather than the test setter.
- Why T keeps its own boundary: `createIncrementalSplit` starts the split at
the *minimum* of all tables' offsets (`:293-337`), and since T's `W_high` is
later than the old tables' offsets the split's `startupOffset` stays where it
was. T's admission boundary therefore comes from its own
`CompletedSnapshotSplitInfo` entry — `shouldEmit` emits only records inside T's
split range with `position.isAfter(highWatermark)`
(`IncrementalSourceStreamFetcher.shouldEmit:292-300`) — plus the
`tableStartupOffsets` gate (`:287-289`). The reader's
`maxSnapshotSplitsHighWatermark` rises to T's `W_high`
(`IncrementalSplitState.java:52-71`), so `markEnterPureIncrementPhaseIfNeed`
keeps T's info until the stream passes it (`:105-117`).
- Why the registry must also carry `W_high`: entering the pure-increment
phase clears `completedSnapshotSplitInfos` in place
(`IncrementalSplitState.java:110-114`), and the `CompletedSnapshotPhaseEvent`
that follows also drops the enumerator's
`assignedSnapshotSplit`/`splitCompletedOffsets` entries for those tables
(`IncrementalSplitAssigner.completedSnapshotPhase:394-408`). After that a
failover rebuilds the split from the registry alone, so admission must also
record `tableStartupOffsets[T] = W_high` — the same per-table floor #13006
added — and `shouldEmit` consults that gate before the split infos
(`IncrementalSourceStreamFetcher.shouldEmit:287-289`).
- Reader-side hand-off, one client at a time: `STREAMING → CLOSING →
SNAPSHOT(T) → BACKFILL(T) → STREAMING(S ∪ {T})`. `CLOSING` is bounded by
`READER_CLOSE_TIMEOUT_SECONDS = 30` then `shutdownNow()`
(`IncrementalSourceStreamFetcher.java:91`, `:247-255`), so the pause is bounded
even if the stream does not stop promptly. `BACKFILL(T)` starts after the close
returns, on the same `database.server.id`
(`MySqlSourceConfigFactory.java:100-102`).
- Resuming the other tables: the enlarged incremental split carries T's
`CompletedSnapshotSplitInfo` (so T starts at `W_high`) and each old table's
`tableStartupOffsets` entry (so it resumes at its own offset instead of
replaying). Both are existing structures after #13006
(`IncrementalSplit.tableStartupOffsets:75`,
`IncrementalSplitState.pendingTableStartupOffsets:91-103`).
## 5. Contract 3 — config-time rejection
- The new option is `scan.binlog.newly-added-table.snapshot.enabled`
(Boolean, default `false`) with
`scan.binlog.newly-added-table.snapshot.interval` (Duration, default 60 s,
lower bound 1 s). The namespace follows #11206's runtime-discovery options; if
#11206 lands under different names, these move with it. Both keys must be
declared in the MySQL factory's `OptionRule`
(`MySqlIncrementalSourceFactory.getOptionRuleBuilder:68`) — undeclared keys are
rejected at config time by `ConfigValidator.validateUnknownKeys` (called from
`ConfigValidationUtils:248`).
- Enabling it requires `exactly_once = true`. This cannot be expressed with
`OptionRule.conditional(...)`: `conditional` only makes the listed options
*required* when the head matches (`OptionRule.java:214-255`, enforced by
`ConfigValidator.checkConditional:462-478`), and it cannot forbid a
combination. The mechanism is a value guard instead, in the shape
`MySqlSpecificOffsetGuards` already uses:
`getOptionRuleBuilder().optional(NEW_OPTION, Conditions.extension(NEW_OPTION,
<guard>))`, where the `ConditionExtension.evaluate(config, value)` reads
`exactly_once` out of the same `ReadonlyConfig` and throws
`OptionValidationException` (see
`MySqlSpecificOffsetGuards.ModeSpecificGuard:76-108` for the pattern of reading
a sibling option and throwing). Note that `exactly_once` defaults to `false`
and `ConfigValidator.hasOption:411-413` goes through `getOptional`, so a guard
must supply the default itself instead of testing presence. Message:
`'scan.binlog.newly-added-table.snapshot.enable
d' requires 'exactly_once = true'.`
- Two further constraints, both also guard-level:
- `scan.binlog.newly-added-table.enabled` must also be `true` (section 1:
otherwise the rest of the pipeline never learns about `T`). Message names both
keys so the user does not have to guess.
- `startup.mode` must be `INITIAL`: that is the only mode that produces
the hybrid assigner with a live snapshot splitter
(`IncrementalSource.restoreEnumerator:378-395`). For MySQL the selectable modes
are `INITIAL`, `EARLIEST`, `LATEST`, `SPECIFIC` and `TIMESTAMP`
(`MySqlIncrementalSourceOptions.STARTUP_MODE:61-74`; `singleChoice` is enforced
by `ConfigValidator.validateSingleChoice:339-366`), and all four of the
non-`INITIAL` ones restore into the incremental-only assigner, which would need
a snapshot branch. Since `INITIAL` is already the default (`:71`), this
restricts nobody's existing config; say if you would rather have the
incremental-only branch in v1. (`SNAPSHOT_ONLY` is not selectable for MySQL at
all — the `SnapshotOnlySplitAssigner` branch at
`IncrementalSource.restoreEnumerator:396-409` is reachable only from the
Postgres/OpenGauss connectors — so no extra rejection is needed for it here.)
- Why not `conditional` for these: the existing `conditional(STARTUP_MODE,
INITIAL, EXACTLY_ONCE)` (`MySqlIncrementalSourceFactory:105-108`) already owns
`exactly_once` under that head; a second conditional with the same head is
rejected by `OptionRule.verifyRequiredOptionDuplicate:484-546`, which is a
further reason to keep the new rules in guards rather than in `conditional`.
- One more constraint that follows from relying on #11206's registration:
`schema-changes.include` must be empty when this mode is on, or the admission
must be rejected. #11206 admits `T` by converting its `CREATE` metadata into a
`CreateTableEvent`, and that event then goes through the ordinary filter;
`SchemaChangeEventType` has no canonical name that maps to
`SCHEMA_CHANGE_CREATE_TABLE` (`SchemaChangeEventType:42-47` lists only
`add.column`, `drop.column`, `modify.column`, `change.column`,
`update.columns`), so any non-empty include list silently excludes the `CREATE`
while the table stays registered and its rows keep flowing. That is #11206's
Issue 3 to fix; until it is, v1 should reject `enabled on + non-empty
schema-changes.include` at config time rather than inherit a half-registered
table. If #11206 lands a different resolution (a canonical `create.table` name,
or admitting only when `CREATE` survives the filter), this bullet becomes
"whatever #11206 decided" and the guard
is dropped.
- OceanBase: `OceanBaseIncrementalSourceFactory` extends the MySQL factory
and only adds `JdbcCommonOptions.COMPATIBLE_MODE` (`:73-79`), so it would
silently inherit the new option. Its only E2E today runs the wrapper against a
MySQL-compatible container with no checkpoint restore
(`OceanBaseCDCCompatibilityIT.testOceanBaseCdcWrapperRuntimeE2e:154`), so v1
adds an explicit rejection there rather than an OceanBase capability. No
OceanBase claim in this revision.
## 6. Backward compatibility
- Option off: the registry is empty, the subtraction reduces to today's
`capturedTables − checkpointCapturedTables`, and no split, state or option
behaviour changes.
- Checkpoints written before this feature restore with empty registry fields
(null-safe).
- `table_pattern` restore semantics and config-added table handling are
unchanged.
- No existing option is renamed or removed; the new options are additive, so
no entry in `docs/en/introduction/concepts/incompatible-changes.md`.
- Hard gate on #11206: the guard from section 5 requires
`scan.binlog.newly-added-table.enabled = true`. On a build without #11206 that
key does not exist, so no valid config can enable this feature. That is
deliberate — unreachable is better than a mode that reads `T` and silently
drops its rows.
- Docs: `docs/en/connectors/source/MySQL-CDC.md` and
`docs/zh/connectors/source/MySQL-CDC.md` get the two options with the
exactly-once precondition and the pause the first admission causes; no
OceanBase doc change in v1.
## 7. Test plan
Unit (`connector-cdc-base`):
1. Registry: `pendingSnapshotTables`, `admittedTables`, `tableWatermarks`
and `tableStartupOffsets` survive a checkpoint/restore round trip; a restored
`pendingSnapshotTables` entry re-drives the split; a second tick and a second
discovery do not admit twice.
2. 2a/2b classification: a split with null watermarks re-runs; a split with
both watermarks is reported finished without a second read (extend
`IncrementalSourceReader`/`SnapshotSplitAssigner` tests).
3. Owner-only dispatch: the enumerator hands the admission split to the
owning reader; after `addSplitsBack` it is re-dispatched to the new owner; a
non-owner reader never receives it.
4. Per-table offsets after admission: extend `IncrementalSplitAssignerTest`
and `IncrementalSourceStreamFetcherTest` so old tables resume at their own
offsets and T starts at `W_high`.
5. Config: enabling without exactly-once is rejected; enabling without
`scan.binlog.newly-added-table.enabled` is rejected; a non-`INITIAL` startup
mode is rejected; enabling with a non-empty `schema-changes.include` is
rejected (section 5); interval bounds are enforced; the OceanBase factory
rejects the option. Extend the existing factory/`ConfigValidator` tests rather
than adding a class.
E2E (MySQL, `connector-cdc-mysql-e2e`):
6. Discovery-off regression: the existing CDC suite, including the
`table_pattern` restore paths, runs unchanged.
7. Crash cuts, deterministic boundaries:
- cut 1: admitted, split assigned, nothing read;
- cut 2a: mid-backfill (the unit re-runs);
- cut 2b: checkpoint carries both watermarks;
- cut 3: after the checkpoint with `admittedTables` + `W_high` + rebuilt
split;
- cut 4: reader reassignment mid-read.
Use the sink trigger (`BEFORE INSERT ... SIGNAL SQLSTATE '45000'`,
already used by `testMysqlCdcContinuesAfterSinkFailure`,
`AbstractMysqlCDCITBase.java:189`/`:245`) to fail exactly once, and
`awaitCompletedCheckpointCountAtLeast` (`MysqlCDCCheckpointRestoreIT.java:397`)
to place the checkpoint boundary instead of sleeps.
8. New conf: `mysqlcdc_to_mysql_with_checkpoint_restore.conf` is
`exactly_once = false`, so the new cases need their own exactly-once conf,
modelled on the module's existing exactly-once variants. Note the sink side
differs from #11206's recommended profile: source `exactly_once = true` here,
and the sink must either be `is_exactly_once = true` or otherwise able to
accept a replayed batch, because the admission backfill can re-emit `T`'s rows
after a 2a re-run. State that explicitly in the docs so a user does not copy
half of #11206's example (which asks for `is_exactly_once = false`).
9. One-client state machine: sample the replication user's `SHOW
PROCESSLIST` server-side and assert there is never a second client with the
same `database.server.id`; run one crash per state in the `STREAMING → CLOSING
→ SNAPSHOT → BACKFILL → STREAMING` machine.
10. Bounded run: repeat the one-client case with `stop.mode = latest` and
assert the job does not finish while an admission backfill is in flight (the
re-arm check in section 4). A separate config case with `startup.mode = latest`
asserts the config-time rejection.
11. End-to-end row visibility: after `T` is created and admitted, assert its
pre-`CREATE TABLE` rows land in the sink. This is the case that fails today
even with a perfect source hand-off (section 1), so it is the real acceptance
test and it pins the dependency on #11206.
OceanBase:
12. No E2E and no capability claim in v1; a unit test asserts the
config-time rejection.
## 8. Open points
1. Option namespace: `scan.binlog.newly-added-table.snapshot.*` is the
proposal, matching #11206's `scan.binlog.newly-added-table.enabled`. If #11206
renames its keys before merging, these rename with it.
2. `startup.mode` scope: `INITIAL` only in v1 versus all modes (needs an
incremental-only snapshot branch).
3. Why the implementation, not the design, waits for #11206. The design can
be agreed now; the code cannot be written usefully before #11206 lands. Three
concrete reasons, so the hold is on the record rather than an unexplained
deferral:
- **A read-side-only implementation produces a job that silently drops
data.** Section 1 is the whole reason: the deserializer returns on an unknown
table
(`SeaTunnelRowDebeziumDeserializeSchema.deserializeDataChangeRecord:248-253`)
and the sink writer throws on one (`MultiTableSinkWriter.write:601-615`). Until
#11206's runtime registration exists there is nowhere to route `T`'s rows, so
an admission snapshot would complete and every row of `T` would be discarded
downstream. Writing the admission state machine against that would mean
developing and testing a feature whose end-to-end path cannot pass.
- **One file collides on the same lines.** #11206 rewrites the
subtraction basis in `IncrementalSource.restore()` (`:435`, guarded by its
`isScanNewlyAddedTableEnabledOnRestore()`), and section 3 extends exactly that
expression. The two are compatible — the registry subtraction nests inside
#11206's `true` branch — but they cannot be two independent PRs editing those
lines; rebasing a half-written admission PR onto a still-moving 49-file branch
is wasted work, not saved work.
- **Review focus.** #11206 is a 49-file change, open since 28 June, with
three blocking findings still open at head `03973e5c`: a two-replica
barrier/monitor deadlock in runtime registration, loss of the checkpointed
runtime table on automatic recovery, and the `schema-changes.include`
interaction that suppresses `CREATE` while still emitting the table's rows. Two
of the three sit in code this feature would extend. Landing both at once would
split review attention across two changes touching the same files, so this
waits until those are settled and then rebases on the result.
- What is *already* enforced mechanically, independent of that policy:
section 5's guard requires `scan.binlog.newly-added-table.enabled = true`, a
key that does not exist on a build without #11206, so no valid config can
enable this feature today. The hold is not the only thing standing between
users and a silent data loss.
4. **The one thing this revision asks of #11206.** Its Issue 2 fix —
restoring a checkpointed runtime table that the submitted DAG catalog does not
know about — is being designed against the assumption that a runtime table only
ever has rows from its `CREATE` position onwards. That assumption stops being
true the moment this feature exists. The specific mechanics on `dev` that make
it visible: `restoreCheckpointProducedType:376-382` keeps only checkpoint
tables already present in the constructed `tables` list and logs `"Ignore
restore table[...] has been deleted."` for the rest, so a table admitted at
runtime is dropped from the deserializer rather than admitted to it;
`historyTableChanges` restore does not replay a `CREATE` either. The fix should
therefore reconcile against the live captured set (or an opt-in eligibility
rule) rather than "the submitted catalog", and should leave room for a
checkpointed table that carries a backfill boundary (`W_high`) in addition to a
`CREAT
E` offset. Nothing in this revision requires #11206 to implement backfill —
only to not foreclose it.
5. Sink profile: whether v1 documents a required sink `is_exactly_once`
value or leaves it to the user with an explicit warning (section 7, item 8).
6. Sequencing once #11206 lands: PR order is registry + option + restore
subtraction first, admission dispatch and reader hand-off second, E2E third.
--
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]
