DanielLeens commented on PR #12134: URL: https://github.com/apache/seatunnel/pull/12134#issuecomment-6538145250
Pushed `NEWHEAD` on top of `6011de825d8`. This round does three things: syncs the branch with `dev`, closes the three test-side blockers from my 10-28 review, and answers @SEZ9's questions on the redrive task and the F2-F8 mapping. Line numbers below refer to `NEWHEAD`. ## 1. Merge with `dev` Merged `origin/dev` `3ac2d1cec6a` (38 commits) with a merge commit, no rebase. One file conflicted, `CheckpointCoordinatorFailoverIT.java`, in two hunks: - Import block: `dev` added `CompletedCheckpoint` (#12042) at the same spot where this PR adds `PendingCheckpoint`. Both imports kept. - `dev`'s new `latestCompletedCheckpoint(...)` helper from #12042 landed at the exact position where this PR inserts its new test method. Both kept: the PR's test first, the `dev` helper right after it (now at `CheckpointCoordinatorFailoverIT.java:1236`). `CoordinatorService.java` and `JobMaster.java` merged without conflict. I compared the PR's pre-merge diff against its merge base with the post-merge diff against `origin/dev` for all three Java files: the added/removed line sets are identical, so the merge neither dropped `dev` code nor changed this PR's production change. The production diff is still the same as in `872701ec8d8`; everything below is test-only. ## 2. Review blockers (all in `CheckpointCoordinatorFailoverIT.java`) **Issue 1 (High): pre-failover guard did not fail fast.** The `Assertions.fail(...)` / `throw new AssertionError(...)` inside the `ignoreExceptions()` poll are gone. The predicate is now side-effect free and returns `true` as soon as either the savepoint is observably mid-shutdown or the client call has already ended (`:1117-1128`). The verdict on a finished call is taken on the test thread immediately after the poll via the new `assertSavepointCallStillPending(...)` (`:1129`, helper at `:1386-1397`): a recorded exception fails the test with that exception as the cause (`"savePointJob failed before master failover"`), a normal return fails it with `"savePointJob completed before master failover"`. The same guard is re-applied immediately before `masterNode1.shutdown()` (`:1138`). `ignoreExceptions()` stays only because the reflective read inside the predicate can race the master's own state transitions; the comment at `:1110-1115` records why the guard must not live inside that la mbda. **Issue 2 (Medium): the test could pass without exercising the redrive.** Two positive assertions were added after `SAVEPOINT_DONE`, both of which fail if `masterNode1` completed the savepoint itself in the check-then-act gap before `shutdown()`: 1. The client call must have failed (`:1188-1193`). `savePointJob` is pinned to `masterNode1`'s member uuid (`SeaTunnelHazelcastClient#requestOnMasterAndGetCompletableFuture`, `JobClient#savePointJob`), so a normal return can only come from the old master answering it, i.e. completing the savepoint. The call's outcome is now recorded in an `AtomicReference<Throwable>` by the caller thread (`:1092-1108`) instead of being swallowed, and asserted non-null after the job is terminal. 2. The savepoint that ended the job must have been triggered strictly after the one observed in flight on `masterNode1` (`:1140-1148` captures the in-flight `PendingCheckpoint`'s trigger timestamp before the kill; `:1201-1221` reads the completed savepoint on `masterNode2` through `dev`'s `latestCompletedCheckpoint(...)` helper, asserts it is a savepoint and that its trigger timestamp is greater). A savepoint completed by the old master would carry the in-flight checkpoint's own timestamp and fail this assertion. I deliberately did not compare checkpoint ids: `CheckpointManager` re-seeds the pipeline's id counter from the latest *completed* checkpoint on restore (`CheckpointManager.java:190`, `StateStoreCheckpointIDCounter#setCount` at `:116`), so the redriven savepoint normally reuses the dropped savepoint's id, which makes the id useless as a discriminator. Both the checkpoint overview store and the checkpoint storage are retained for `SAVEPOINT_DONE` (`CheckpointManager#clearCh eckpointIfNeed` and `CoordinatorService#cleanupTerminalZombieCheckpointIfNecessary` only clean up for `FINISHED`/`CANCELED`), and the overview is updated synchronously on completion (`CheckpointCoordinator.java:1449-1452` -> `HazelcastCheckpointOverviewStateStore#updateOverview`, an `IMap#compute`), so the read is not racy. To support (2) the two white-box helpers now return the in-flight `PendingCheckpoint` instead of a boolean (`findInFlightSavepointCheckpoint` `:1358`, `findUnacknowledgedSavepointCheckpoint` `:1414`); their logic is unchanged. **Issue 3 (Medium): multi-pipeline behavior.** This is a source trace, not a test, and I want to be explicit about that. I did not add a two-pipeline E2E because the partial-completion shape (one pipeline's savepoint already complete, another still pending, then a master kill) cannot be produced deterministically with the current fixtures: it needs one pipeline's barrier acknowledgement held back while a sibling's completes, and there is no hook for that. - The redrive has no per-pipeline logic. It calls the job-level `JobMaster#savePoint()` (`CoordinatorService.java:1277`), which fans out to every coordinator through `CheckpointManager#triggerSavePointsAndWaitComplete` (`CheckpointManager.java:263-267`). For N pipelines that were all still in flight at failover, each is restored `RUNNING`, each `reportedPipelineRunning` eventually flips its `isAllTaskReady`, and the next attempt completes all of them, exactly as in the single-pipeline case. - Partial completion (pipeline A already `FINISHED` through its savepoint, B still in flight): A's coordinator is rebuilt with the completed savepoint as a *restored* latest checkpoint (`CheckpointCoordinator.java:268-270`), so `isCompleted()` is `false` and, since `SubPlan#restorePipelineState` only calls `reportedPipelineRunning` for a `RUNNING` pipeline (`SubPlan.java:649-657`), its `isAllTaskReady` stays `false`. Every attempt therefore gets `TASK_NOT_ALL_READY_WHEN_SAVEPOINT` from A (`CheckpointCoordinator.java:922-924`). The claim in my review that this spins for 5 minutes is wrong under the `savePoint()` that `dev` has carried since #11489: as long as B is also not ready, every pipeline fails with a precondition reason, `isSavepointStartPreconditionFailure` is `true` (`JobMaster.java:1530-1540`, both `TASK_NOT_ALL_READY_WHEN_SAVEPOINT` and `CHECKPOINT_COORDINATOR_SHUTDOWN` count, `:1546-1556`), the job is put back to `RUNNING` and the redrive retries 500 ms later, the same transient as the single-pipeline case. The first attempt in which B is ready starts and completes B's savepoint (`SUSPEND`); `waitSavepointCompleted` then reports `anyPipelineSuspended == true`, so the result is no longer classified as a precondition failure and `savePoint()`'s `finally` calls `PhysicalPlan#savepointFailed()` (`JobMaster.java:1437-1453`) -> `stopJob()` -> `stopPipelineWithCheckpointFallback` for every pipeline (`PhysicalPlan.java:238-260`, `SubPlan.java:603-617`). B's coordinator `isNoErrorCompleted()` is `true`, so B is force-finished. A takes the `cancelCheckpointCoordinator()` + `forceStopPipeline()` branch because `isNoErrorCompleted()` excludes restored checkpoints (`CheckpointCoordinator.java:1510-1520`), but `forceStopPipeline` is a no-op for end-state vertices (`PhysicalVertex.java:586-590`) and A's pipeline future was already completed as `FINISHED` at restore (`SubPlan.java:837-842`), so the job's finished/failed/canceled counting is unchanged. With all pi pelines finished, none failed or canceled, and the job still `DOING_SAVEPOINT`, `PhysicalPlan` resolves it to `SAVEPOINT_DONE` (`PhysicalPlan.java:172-186`). The redrive loop sees the terminal state on its next tick and exits (`CoordinatorService.java:1265-1267`). This is the same fallback path a non-failover savepoint takes on `dev` when one pipeline of a job has already finished before the savepoint is requested (A's coordinator then answers `CHECKPOINT_COORDINATOR_SHUTDOWN` instead, which is classified identically). - What this does cost: in the partial-completion case the redriven `savePoint()` itself completes exceptionally (the first exception is A's precondition rejection), which the loop logs at INFO as "did not complete ... retrying" even though the job is already on its way to `SAVEPOINT_DONE` (`:1290-1297`). That is the logging accuracy point from Issue 5 (non-blocking); the outcome of the job is not affected. I did not touch the PR description in this round; the multi-pipeline result above and the stale "widened assertion" note (Issue 6) still need to be folded into it. ## 3. @SEZ9: the redrive task body and its termination conditions `CoordinatorService#redriveSavepointAfterMasterFailover` (`CoordinatorService.java:1256-1324`), called from the scheduling lambda at `:517-518` right after `consumeRedriveSavepointAfterRestore()` and before `jobMaster.run()`. It submits one task to the job's MDC-traced executor; the task is: - `deadline = now + REDRIVE_SAVEPOINT_WAIT_MILLIS` (5 minutes, `:142`, `:1262`); `while (now < deadline)` (`:1263`). - Read `jobMaster.getPhysicalPlan().getJobStatus()` (`:1264`). If it is an end state, return (`:1265-1267`). If it is `RUNNING`, call `jobMaster.savePoint().get()` (`:1268-1297`): `true` returns; `false` or any exception is logged at INFO and the loop falls through to the sleep. Any other status (`PENDING`, `SCHEDULED`, `DOING_SAVEPOINT`, ...) just falls through to the sleep. `Thread.sleep(500)` (`:1300`); an interrupt anywhere restores the flag and returns (`:1288-1290`, `:1301-1304`). - On deadline exhaustion: one WARN that the client's original savepoint request will need to be retried manually (`:1306-1313`), and the task ends. It never changes job state itself. Your two cases: - *The job never reaches `RUNNING`*: the loop only polls (500 ms reads of a volatile status) for 5 minutes and then logs the WARN. It never calls `savePoint()`, never touches the job; `jobMaster.run()` is unaffected because it does not depend on this task. If the job reaches a terminal state in the meantime, the loop exits on that instead. - *`savePoint()` keeps failing*: there are only two kinds of failure, and `JobMaster#savePoint()`'s own `finally` (`JobMaster.java:1437-1453`) handles both before the redrive sees the result. (a) Every pipeline rejected the request for a precondition reason (`TASK_NOT_ALL_READY_WHEN_SAVEPOINT`, `CHECKPOINT_COORDINATOR_SHUTDOWN`): the job is put back to `RUNNING` via `restoreRunningAfterSavepointStartFailure()` (`:1469-1473`) and the redrive retries on its next tick. This can only repeat while tasks are still redeploying; once every task has reported `READY_START` the precondition holds. If it somehow never holds, the 5-minute deadline ends the loop. (b) Anything else: `PhysicalPlan#savepointFailed()` stops every pipeline through the checkpoint fallback (`PhysicalPlan.java:238-260`) and the job reaches a terminal state (`SAVEPOINT_DONE` if all pipelines had completed their savepoint, otherwise `CANCELED`/`FAILED`), which the redrive sees as an end state and exits. So a non-precondi tion failure cannot repeat: the first one drives the job terminal. There is no path that leaves the job parked in `DOING_SAVEPOINT`. - Two limits, both documented in the Javadoc: the deadline is checked between attempts, so one `savePoint().get()` is bounded by the savepoint's own completion/timeout rather than by the 5-minute budget; and if the executor rejects the submit, the redrive is skipped with a WARN (`:1315-1324`), the job is unaffected and the request stays dropped as before this fix. ## 4. F2-F8 against the current code - **F2 / F5** (savepoint still pending when the old master goes down): before `shutdown()` the test re-checks that the client call is still blocked (`:1138`) and that a savepoint `PendingCheckpoint` is still unacknowledged (`:1140-1147`); after `SAVEPOINT_DONE` it proves the redrive produced the result, not the old master: the client call failed (`:1188-1193`) and the completed savepoint was triggered after the in-flight one (`:1201-1221`). A savepoint that finished on the old master now fails the test instead of passing it. - **F3** (reflection throwing mid-transition): the 10 ms poll keeps `ignoreExceptions()` (`:1120`), but nothing inside the predicate throws on purpose any more; the fail-fast was moved out of it for the reason above. - **F4 / F7 / F8** (executor and future): `savepointCaller.shutdownNow()` is the first statement of `finally` (`:1223`), before closing the client and both masters. The future's outcome is inspected three times: after the poll (`:1129`), immediately before `shutdown()` (`:1138`), and after the job is terminal (`:1188-1193`), where the recorded failure is asserted rather than logged. A pre-failover failure surfaces directly with its cause, not as a 30 s timeout. - **F6** (`pendingCheckpoints` reflection): still reflected via `ReflectionUtils.getField` (`:1414-1429`); the Javadoc at `:1399-1412` explains the coupling and why `isFullyAcknowledged()` cannot be used. No production accessor was added to keep the production diff unchanged. ## 5. Verification Per the validation policy for this repository only formatting was run locally (`spotless:apply` / `spotless:check` on `connector-seatunnel-e2e-base`); compile and tests are left to CI. Fork `Build` run for `NEWHEAD`: RUNLINK. I have not looked at its result and make no claim about it here. -- 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]
