DanielLeens commented on PR #11382:
URL: https://github.com/apache/seatunnel/pull/11382#issuecomment-5370826416
@SEZ9 Thanks for the deep pass. I pulled the raw source for the current head
(`b9cfa1a08550`, confirmed no new commit since my last review) directly from
the fork and re-traced this myself rather than just re-asserting my earlier
conclusion, since Issue 2 is a direct contradiction of something I'd already
traced end-to-end six hours earlier.
**On Issue 2 ("no re-delivery path after failover"): the call site exists,
and it's the same one I cited in my last review.**
`CheckpointCoordinator.notifyCheckpointCompleted(CompletedCheckpoint)` is
not "fired once at original completion" only — it's also invoked from the
restart/failover path:
```text
CheckpointCoordinator.reportedTask(TaskReportStatusOperation)
[CheckpointCoordinator.java:309-319]
-> case READY_START: allTaskReady()
CheckpointCoordinator.allTaskReady()
[CheckpointCoordinator.java:438-463]
-> isAllTaskReady.compareAndSet(false, true)
[line 447]
-> notifyCompleted(latestCompletedCheckpoint)
[line 453]
CheckpointCoordinator.notifyCompleted(CompletedCheckpoint)
[CheckpointCoordinator.java:466-490]
-> notifyCheckpointCompleted(completedCheckpoint)
[line 475]
-> new CheckpointFinishedOperation(taskLocation, checkpointId, true,
checkpoint.getCheckpointType())
sent to EVERY task in plan.getPipelineSubtasks(), including a
brand-new sink [lines 1358-1369]
```
The reason this actually fires again after a failover restart, not just
once: `isAllTaskReady` gets reset to `false` on every coordinator reset —
- `cleanPendingCheckpoint(CheckpointCloseReason)` unconditionally does
`isAllTaskReady.set(false)` (`CheckpointCoordinator.java:1130`), which runs on
failure/restart cleanup, and
- the other restart branch does the same when tasks haven't already started
(`CheckpointCoordinator.java:707`, the `else` of the `alreadyStarted` check at
line 695).
So on the next restart, every restored task reports `READY_START` again,
`allTaskReady()`'s `compareAndSet(false, true)` succeeds again, and
`notifyCompleted(latestCompletedCheckpoint)` replays.
`latestCompletedCheckpoint` is the *persisted* checkpoint deserialized in the
constructor (`this.latestCompletedCheckpoint = ...;
latestCompletedCheckpoint.setRestored(true)`), so its `checkpointType` is
whatever was last durably completed — in the exact
failover-between-schema-change-checkpoints scenario you and I are both
describing, that's the schema-change-before checkpoint. The freshly-constructed
`SchemaChangeDrainGuard` on the restored sink receives this typed
`CheckpointFinishedOperation`, and `checkpointCompleted(checkpointId,
checkpointType)` opens the window via `isSchemaChangeBeforeCheckpoint` matching
on type alone (`SchemaChangeDrainGuard.java:105-113, 173-178`) — no dependency
on the new task having seen the id before.
I'd genuinely like to know if you were looking at an older revision of the
guard (pre the `checkpointType`-based, id-agnostic match) — if so that'd
explain the gap you're describing, since that's exactly the shape of the fix I
traced. Happy to walk through it further if you're still seeing a hole in this
specific chain.
**On Issues 1 and 3: these aren't something I'd previously ruled on, so I'm
not brushing them aside.**
- Issue 3 (`writeInternal` NPEs if `checkpointType` is null): confirmed the
code shape — `out.writeString(checkpointType.getName())` at
`CheckpointFinishedOperation.java:82` has no null guard. I only found one
production call site (`CheckpointCoordinator.java:1365-1369`), and it always
passes `checkpoint.getCheckpointType()` off a `CompletedCheckpoint`, which is a
required, always-set field there — so it isn't reachable today, but I agree
defensive null-handling on both `writeInternal`/`readInternal` is a reasonable
hardening ask regardless.
- Issue 1 (notification-vs-data race for the *non-restore* path): this is a
distinct claim from Issue 2 — about ordering between the source's and sink's
own async completion notifications versus the in-band `SchemaChangeEvent`, not
about restore. I haven't verified yet whether this race predates this PR (the
same id-based `checkpointBarrierHandled` + `checkpointCompleted(id)` path
existed before this diff) or is newly exposed by it. I want to check that
properly rather than give you a quick take on something this deep in the
checkpoint protocol.
Given there's no new commit here, let's not turn this into a second full
re-review pass over comment threads — once a revision addresses Issue 1/3 (or
clarifies why they don't apply), I'll do a fresh full re-review of the actual
diff, including verifying Issue 1 against the pre-PR baseline at that point.
For now, from my side the restore/failover gap (Issue 2) is closed on the
current head; Issues 1 and 3 are fair asks and worth folding into that next
pass.
--
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]
