ryanmeowy commented on issue #10203: URL: https://github.com/apache/seatunnel/issues/10203#issuecomment-6379487210
@SEZ9 both answers, checked against `dev`: **Row 2b: what makes the restored split read as complete.** The checkpointed value is a `SnapshotSplitState`, which carries exactly two fields, `lowWatermark` and `highWatermark` (`SnapshotSplitState.java:31-32`), and `toSourceSplit()` writes them back onto the `SnapshotSplit` (`:39-49`). The predicate the restore path uses is `SnapshotSplit.isSnapshotReadFinished()` = `lowWatermark != null && highWatermark != null` (`SnapshotSplit.java:70-72`). Both fields are set by the record emitter as it consumes the watermark records, not by the read task: `IncrementalSourceRecordEmitter.processElement()` calls `setLowWatermark()` on the LOW watermark record and `setHighWatermark()` on the HIGH one (`:133-139`, setters at `:136` and `:138`). Null on both is the only other state a snapshot split has, so "different from pending" is exactly "high watermark recorded". One caveat that matters for the admission design, and that I had wrong in my own note until I traced it: the HIGH watermark is dispatched by `MySqlSnapshotSplitReadTask` right after the table data is read and **before** the bounded backfill starts (`setHighWatermark` at `:154`, dispatch at `:155-159`), with the backfill running afterwards inside `MySqlSnapshotFetchTask.execute()` (`:95-116`). So "high is set" means the snapshot rows are read, not that the backfill is done. That is still safe, but the reason is a different mechanism than the watermark: the split only enters `finishedSplits()` once the whole fetch task has returned, `IncrementalSourceScanFetcher.isFinished()` requires `!taskRunning && !hasNextElement && reachEnd` (`:109-112`), and only then does `SourceReaderBase.finishCurrentFetch()` call `onSplitFinished()` and move the split into `finishedUnackedSplits` (`SourceReaderBase.java:199-214`, `IncrementalSourceReader.java:200-203`). A death during the backfill therefor e leaves the split out of that map and re-runs the unit, which is the 2a path — the persisted high watermark alone does not promote it to 2b. I will state that split explicitly in the revision rather than relying on "the state says finished". **Row 3: per table, not global.** `shouldEmit()` resolves `tableId` per record (`IncrementalSourceStreamFetcher.java:268`) and the comparison is `maxSplitHighWatermarkMap.get(tableId)`, a per-table map built by `configureFilter()` from each `CompletedSnapshotSplitInfo` (`:315-345`, the map is assigned at `:344`). There is no global watermark anywhere in that path. The one refinement: once a table reaches `position >= maxSplitHighWatermarkMap[tableId]`, `hasEnterPureBinlogPhase()` adds it to `pureBinlogPhaseTables` and later records for that table skip the per-record check entirely (`:301-313`). The switch point is the same `W_high`, so "drops `< W_high`" holds semantically, but the implementation is "filter per record until the table crosses its own watermark, then emit unconditionally for that table". I will word row 3 that way instead of implying a per-record comparison forever. Nothing else changes; the acceptance contract, the deterministic-cut test plan and the MySQL/OceanBase split from your previous comment are what the revision will be organised around. -- 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]
