ryanmeowy commented on issue #10203:
URL: https://github.com/apache/seatunnel/issues/10203#issuecomment-6357660652
@DanielLeens @SEZ9 the transition table for the four cuts. One question per
row: after the crash, how many times does T's snapshot run, and what happens to
the rows it already produced. T is the table admitted while the job was running.
Terms mapped to your wording from the 16th: admitted = `admittedTables`,
in-flight = `pendingSnapshotTables`, watermark = `tableWatermarks[T] = W_high`.
Two rules it depends on:
- The enumerator is the only writer of the registry. A reader reports a
finished split and never admits a table, so a reader crash or reassignment
cannot produce a second admission; a poller tick that sees T already in
`pendingSnapshotTables` is a no-op, which covers the timer-tick duplicate.
- T joins the incremental split's `tableIds` only in a checkpoint that also
carries `admittedTables[T]` and `tableWatermarks[T]`. Anything before a
completed checkpoint is lost on recovery, so a cut before one re-runs instead
of resuming.
| # | cut | durable when it dies | recovery | T's snapshot | T's rows |
| - | --- | --- | --- | --- | --- |
| 1 | admitted, split assigned, nothing read yet | registry has no T, or
`pendingSnapshotTables[T]` if a checkpoint landed after the decision | the
poller re-discovers T on its next tick, or point 6's re-drive recreates the
split from `pendingSnapshotTables` | once, from a fresh `[low', high']` | none
produced |
| 2 | `low`/`high` captured and the backfill done, but the checkpoint
carrying `W_high` never completed | `pendingSnapshotTables[T]`; the reader
state holds the split only if a checkpoint landed after the assignment | the
split is handed back (2a / 2b below), or point 6's re-drive recreates it if the
checkpoint predates the assignment | 2a re-runs, 2b does not | 2a replays the
split |
| 3 | checkpointed: `admittedTables[T]` + `W_high` + the rebuilt incremental
split | all three, in the same checkpoint | nothing to drive: restore subtracts
the registry, and the incremental branch has no pending entry | does not run |
`shouldEmit()` drops `< W_high` |
| 4 | the reader holding the split is reassigned mid-read | unchanged;
whatever the last checkpoint carried | the split goes back through
`addSplitsBack` and is dispatched to the reader that holds the streaming split
| as row 1 or row 2 | as row 1 or row 2 |
**Death inside the unit.** @SEZ9 asked on the 11th what is persisted if the
job dies after the snapshot read but before the backfill finishes. Nothing
about `[low, high]`, and that is deliberate: `MySqlSnapshotFetchTask.execute`
runs the snapshot read and the bounded backfill as one task, and the split only
counts as finished once that task has returned and drained its queue
(`IncrementalSourceScanFetcher.isFinished()`), so a death between the two is 2a
— the pair re-runs with a fresh range. The range never needs persisting because
it is never half-applied; what must not happen is a second admission, and that
is the registry's job.
**2a / 2b** differ in one bit: whether the last completed checkpoint already
contains the finished split. If it does not (2a), the split comes back
unfinished and re-runs from a fresh `[low', high']`. If it does (2b), the
watermark is recovered without replaying: `SnapshotSplitAssigner.addSplits`
calls `restoreCompletedSnapshotSplit`, which — only when the split is
`isSnapshotReadFinished()` and still present in `assignedSplits` — puts its
`SnapshotSplitWatermark` back into `splitCompletedOffsets` and drops the split
instead of queueing it. The reader side cooperates: since #12163
`finishedUnackedSplits` is cleared only on the enumerator's
`CompletedSnapshotSplitsAckEvent`, and `snapshotState()` writes those splits
into the checkpoint, so a reader preempted before the ack restores them as
finished and re-reports on `addSplits`. That precondition is the one to respect
when we add T: the admission split must be registered in `assignedSplits` for
the whole read, or 2b silently de
grades to 2a.
**Correction to my last note.** I wrote there that the re-read happens "with
nothing yet committed". That only holds if no checkpoint completed while the
snapshot was being read. If one did, the rows it carried are committed and the
2a re-read replays them, because a split read is not resumable:
`SnapshotSplitState` holds only the two watermarks, and both are set when the
read ends. That is the exposure the initial snapshot phase already has for
every table today, not something admission introduces, and it is not what the
exactly-once precondition in point 1 is for. It does mean the cut-2 test
asserts "no loss and no second admission", not "no replay at all", unless the
sink keys on the primary key.
**Cut 4** needs one rule the note left implicit. `assignSplits()` offers
`getNext()` to whichever reader is waiting, and after #11677 the loop re-runs
when splits come back, so the returned snapshot split can be picked up by a
reader that does not hold T's streaming split. If that happens the split sits
in that reader's queue forever: `checkSplitOrStartNext()` returns early while a
stream fetcher is alive (`IncrementalSourceSplitReader.java:153-156`), and an
unbounded stream fetcher never finishes. So the enumerator must hand T's
admission split only to the reader that holds S, and the incremental-phase
assigner needs a snapshot branch for it: `IncrementalSplitAssigner.addSplits`
still casts every returned split to `IncrementalSplit`, and `getNext()` keeps
returning empty once `splitAssigned` is set. With that, cut 4 is row 1 or row 2
on another subtask, and the returned split keeps its `splitId`, so dedup does
not depend on which reader picks it up.
**The hand-off, for the state-machine test.** Point 2's stop/restart shape,
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:89`,
`:246-251`); `BACKFILL` starts only after it returns, on the same
`database.server.id`. So the test asserts one thing, never two clients with the
same server id connected (sampled server-side from the replication user's `SHOW
PROCESSLIST`), plus one crash per state that has to land in row 1 or row 2.
There is no reserved-id branch, hence one path, not two.
**Tests.** Rows 1 and 2a end in the same place, so one case covers both:
give T a row the sink rejects (a sink trigger as in #11677, or a value the sink
column cannot hold), and the crash is caused by the unit itself, so it always
lands inside it. Whether any of T's rows were emitted before that only decides
between row 1 and 2a, not what recovery does. The assertion is T's full history
plus every change after admission, exactly once. Row 3 needs the checkpoint to
be in first, and the harness already has the two pieces:
`awaitCompletedCheckpointCountAtLeast()` for the barrier and the `BEFORE INSERT
... SIGNAL SQLSTATE '45000'` trigger on the sink table used by #11677's
`testMysqlCdcContinuesAfterSinkFailure` to force the failure exactly once,
confirmed through the engine log rather than a sleep. Row 2b cannot be produced
deterministically from outside and is a reader unit test instead: `addSplits`
with a finished split must re-report it and the enumerator must admit without a
seco
nd read. Row 4 is row 1 or 2 on another subtask, so it is the same cases run
with parallelism above 1.
One constraint that comes out of that:
`mysqlcdc_to_mysql_with_checkpoint_restore.conf` is `exactly_once = false` on
both sides and this feature requires exactly-once, so the new cases need their
own conf (the module already has exactly-once variants to copy from). With
discovery defaulting off, the existing CDC suite is the `table-pattern`
backward-compatibility regression; I will still add one case that asserts a
discovery-off job's restore path is unchanged.
--
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]
