DanielLeens commented on PR #11382: URL: https://github.com/apache/seatunnel/pull/11382#issuecomment-5645087530
@SEZ9 on F1 specifically, since you asked me to confirm the wait/notify shape rather than just the fact that it's `synchronized`: I traced it myself against `1ee22aced026` and it holds up on all three points that actually matter for this pattern, not just the surface-level "synchronized + notifyAll." - **Missed-notification race**: both `checkpointCompleted(long, CheckpointType)` and `checkSchemaChangeCanApply(...)` are `synchronized` on the same monitor (`this`). So there are only two possible orderings: either the waiter is already parked in `wait()` (which atomically releases the monitor) when the notifier acquires the lock, sets `schemaChangeDrainReady = true`, and calls `notifyAll()` — standard wake-up; or the notifier runs first and sets the flag before the waiter ever enters `checkSchemaChangeCanApply`, in which case the waiter's `while (!schemaChangeDrainReady)` check simply evaluates false and it never blocks. There's no window between "check condition" and "start waiting" where a notification could slip through unseen, because both operations are mutually exclusive under the same lock. - **Spurious wakeups**: `wait(remainingMillis)` sits inside a `while` loop, not an `if`, so a spurious return re-evaluates `schemaChangeDrainReady` and either exits (flag now true) or recomputes `remainingMillis` and waits again. It doesn't trust a bare wakeup as proof of completion. - **Condition re-check / timeout budget**: `deadline` is computed once up front and `remainingMillis` is recomputed from it on every loop iteration, so a sequence of spurious wakeups can't reset or extend the effective 30s budget — it correctly degrades to `remainingMillis <= 0` and throws once the real deadline passes, rather than waiting 30s again per spurious wakeup. Threading-wise this also lines up with why the guard needs to be a monitor at all: `checkSchemaChangeCanApply` runs on the sink task's own execution thread (`SinkFlowLifeCycle.processSchemaChangeEvent`), while `checkpointCompleted` is driven by `notifyCheckpointComplete`, which arrives via the coordinator's completion fan-out on a different thread. So this is a genuine cross-thread producer/consumer signal, not a same-thread call where `wait`/`notify` would be pointless. I don't see a missed-notification path, and F1 is resolved as designed. On the rest, all of this was already answered directly in my previous comment against this same head, so pointing back rather than re-deriving it: - **F2**: still open — the `restoreCoordinator(false)` mechanism you cited is correct, but the javadoc on `checkpointCompleted` still asserts the coordinator is recreated on restart, which I confirmed is not true (`SubPlan.prepareRestorePipeline()` looks the existing coordinator up rather than reconstructing it). Needs the javadoc fixed to describe `isAllTaskReady.set(false)` as the actual re-arming mechanism. - **F3/F8**: confirmed still half-open. I re-checked `CheckpointFinishedOperation.readInternal` (`CheckpointFinishedOperation.java:99`) and `CheckpointType.fromName` myself just now — `fromName` throws an unchecked `IllegalArgumentException` with no catch anywhere on the read path, so an unrecognized type name on the wire (e.g. version skew during a rolling deploy) still blows up the operation thread. Write side's `IOException` fail-fast is fine as-is. - **F4**: still open, tied to F2 — yes, `schemaChangeDrainReady` is still guard-local process memory with no snapshot/state involved; the coordinator's replay of `latestCompletedCheckpoint` is what substitutes for that on recovery, but that dependency needs to be stated in the javadoc, not just established in this thread. - **F5**: resolved — `SinkFlowLifeCycle.notifyCheckpointAborted(long, CheckpointType)` (`SinkFlowLifeCycle.java:356`) does call `schemaChangeDrainGuard.checkpointAborted(checkpointId, checkpointType)`, so there's a real coordinator-driven sender now, not an unreachable cleanup method. - **F6**: still open, unchanged — `schemaChangeBeforeCheckpointId` is still a single `long`, so two overlapping schema-change-before/after checkpoint pairs still can't be distinguished by the guard. - **F7**: still open — current tests exercise the guard directly or stub `notifyCompleted`; nothing goes through the real `SinkFlowLifeCycle` -> `SeaTunnelTask` -> `CheckpointFinishedOperation` serialization/dispatch round trip, and there's no e2e schema-evolution-with-restart scenario in this diff. So, unchanged bottom line: F1 and F5 resolved (F1 now independently re-verified at the wait/notify level, not just the synchronized-keyword level), F2/F4 open together pending the javadoc fix, F3/F8 half-open pending defensive handling of an unrecognized wire-format type name, F6/F7 open and unchanged. Same head, same gate — happy to re-review the moment a commit lands. -- 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]
