DanielLeens commented on PR #11382: URL: https://github.com/apache/seatunnel/pull/11382#issuecomment-5564078629
@SEZ9 thanks for narrowing this down and confirming F1 (the notification race) is resolved. On F2 (the restore-loop concern), I want to push back with the specific mechanism rather than just repeat my earlier note — I re-traced it end-to-end from source at this exact head (`634a0325a023`) before answering: `CheckpointCoordinator.allTaskReady()` (`CheckpointCoordinator.java:438-463`) fires once every subtask in the pipeline reports `READY_START` — which is exactly what happens after a restore, once the freshly-restored sink task comes back up. It unconditionally calls `notifyCompleted(latestCompletedCheckpoint)` (`:453`), and `latestCompletedCheckpoint` is populated straight from the persisted `PipelineState` in the coordinator's constructor (`:252-255`, `setRestored(true)`) whenever the coordinator itself is recreated after a failover — that part is pre-existing code, untouched by this PR. `notifyCompleted` -> `notifyCheckpointCompleted(completedCheckpoint)` (`:1358-1372`) then iterates `plan.getPipelineSubtasks()` — every task location in the pipeline, source and sink alike — and sends each one a `CheckpointFinishedOperation` carrying `checkpoint.getCheckpointType()`. This isn't scoped to "tasks that already saw the barrier"; it goes to every subtask, including a brand-new restored sink whose `SchemaChangeDrainGuard` starts at `schemaChangeDrainReady=false`. On the receiving side, `SchemaChangeDrainGuard.checkpointCompleted(id, type)` (`SchemaChangeDrainGuard.java:133-144`) delegates to `isSchemaChangeBeforeCheckpoint(id, type)` (`:225-230`), which for a non-null type checks only `checkpointType.isSchemaChangeBeforeCheckpoint()` — it does not require the id to match anything the fresh guard has tracked locally. So the replayed BEFORE-type notification opens the freshly-restored guard's window unconditionally. Concretely: in the exact race window your writeup describes, `latestCompletedCheckpoint` at restore time is the schema-change-before checkpoint (no later checkpoint has completed yet), so the first time all tasks become `READY_START` after recovery, this replay fires and the restored sink's guard opens before the source's re-emitted `SchemaChangeEvent` can reach it. I don't think there's a re-delivery gap here — `allTaskReady()` is the re-delivery path, it's just pre-existing machinery this PR reuses rather than something it adds, which is easy to miss reading only the diff. If you see a concrete path where `allTaskReady()`/`notifyCompleted()` would *not* fire for a freshly-restored sink, or where `latestCompletedCheckpoint` could be stale/null in this exact scenario, I'd genuinely like to see it — that's the one thing that would change my read here. On F3 (wire format / unknown type name): I'm keeping this non-blocking, but I don't think it's fully closed either. The null case is now handled cleanly, and I agree `CheckpointType.fromName` still has no fallback for a name it doesn't recognize on the read side. I'm not treating that as a merge blocker because (a) there is exactly one production call site and it always supplies a concrete, currently-valid type, and (b) I'm not aware of an existing guarantee elsewhere in Zeta's internal Hazelcast operation surface for surviving a mixed-version rolling restart mid-cluster — this PR isn't introducing a new category of risk relative to how the rest of the checkpoint RPC already behaves. That said, it's a small, low-risk defensive addition (fall back to the untyped path on an unrecognized name), so worth doing in a follow-up regardless of blocking status. Separately, on CI: I dereferenced the apache-side `Build` pointer to the real fork run (`davidzollo/seatunnel` run `34012458585`). Every lane that actually exercises this diff passed, including both `engine-v2-it (8/11)` and all four `unit-test` shards. The run's overall `cancelled` conclusion traces to one unrelated job, `kudu-connector-it (8, ubuntu-latest)`, which ran for roughly 90 minutes before being cancelled — Kudu isn't touched by any of this PR's seven files, so this looks like an infra/timeout cancellation rather than a regression from this diff. @davidzollo, a rerun of just that job (or the workflow) should clear it without needing a code change. To be clear about where this leaves things: I don't consider F2 or the residual F3 gap to be merge blockers given the above, but I'd like your read on the restore-path trace specifically before either of us calls it settled — that's the one point where we currently disagree on the facts, not just on severity. -- 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]
