Vivek1106-04 commented on PR #12165: URL: https://github.com/apache/seatunnel/pull/12165#issuecomment-6435747333
@DanielLeens thanks for the full re-read, and for catching the expiry hand-off. You are right: once the watchdog has handed the parameterless runnable to the coordinator executor, `cancelAll()` cannot reach it, and `handleCoordinatorError` only checks the *current* future. Both items are addressed at head `a932e27a7`. **Issue 5 (stale expiry fails a restored coordinator)** - `3cc4dea45` - `startTriggerPendingCheckpoint` reads `checkpointCoordinatorFuture` once, while its caller still holds `lock` and before the checkpoint exists. That future is the generation token. The watchdog now calls `expireCheckpoint(pendingCheckpoint, owner)`. - The submitted action, and the inline rejection fallback, both run `failExpiredCheckpoint(expired, owner)`. Under a new `generationLock` it checks `owner == checkpointCoordinatorFuture`, `!owner.isDone()` and `pendingCheckpoints.get(id) == expired`. Only if all three hold does it set FAILED, complete `owner` and clean the pending state. Otherwise it logs at INFO and drops the expiry. - `restoreCoordinator` swaps `errorByPhysicalVertex` / `checkpointCoordinatorFuture` and writes RUNNING under the same `generationLock`. So the expiry either fails the old generation before the swap, or sees the new future and is dropped. `checkpointCoordinatorFuture` is now `volatile`, because the swap and the token read no longer share a monitor. - Why a new monitor instead of `lock`: `startSavepoint` holds `lock` in its sleep-poll until `pendingCounter` drops, and that can be the checkpoint that is expiring. Taking `lock` here would deadlock. Taking `generationLock` does not: `cleanPendingCheckpoint` sets `shutdown = true` before it takes `lock`, which ends that poll, same as today. - `checkpointManager.handleCheckpointError(...)` runs after `generationLock` is released. It takes the `SubPlan` monitor, and a restoring pipeline holds that monitor when it calls `restoreCoordinator`, so calling it under the lock would invert the order. Because of this, the expiry path now cleans pending state before notifying the manager instead of after. The old comment's reason for that order (do not kill the pool before the manager is notified) no longer applies: `cleanPendingCheckpoint` only calls `cancelAll()`, which cancels timers with `cancel(false)` and never interrupts. I checked the sync consumers of the futures completed under the lock. `SubPlan` and `JobMaster` block with `join()`/`get()`, and the pending-checkpoint handlers are `*Async`, so no callback takes the pipeline monitor under `generationLock`. - The other `handleCoordinatorError` callers are unchanged. They run synchronously in the generation that raised the error, not as a deferred action. Tests in `CheckpointCoordinatorTest`: - `testExpiryHeldAcrossCoordinatorRestoreDoesNotFailTheRestoredGeneration`: a single-thread coordinator executor is held by a latch, so the expiry queues behind it. The test then calls `restoreCoordinator(false)` and puts the *same* `PendingCheckpoint` back into `pendingCheckpoints`, so only the generation check can drop it. Then it releases the latch and asserts the restored future is not done and `handleCheckpointError` is never called. - `testExpiryOfTheCurrentGenerationFailsTheCoordinator`: the same expiry without a restore still completes the future FAILED and calls `handleCheckpointError(1, false)`. The existing `testCheckpointExpiryHandlingNeverRunsOnTheDispatchThread` still covers the end-to-end watchdog path. Red runs, made by changing only the production code: - With the action reverted to the old unconditional `handleCoordinatorError(CHECKPOINT_EXPIRED, null)`, the held-expiry test fails: `an expiry of the generation before the restore must not fail the restored one ==> expected: <false> but was: <true>`. - With only the `owner` check removed (pending-identity check kept), it fails the same way. So the token is what does the work, not the map lookup. **Issue 6 (thread state is not task settlement)** - `a932e27a7` `testFailingBodyCompletesExceptionallyAndKeepsItsDispatchThread` now waits for `lease.outstandingCount() == 0` first, so the task has passed its `finally`. Then it waits for the worker to reach `TIMED_WAITING` or `TERMINATED` and asserts `TIMED_WAITING`. After `finally`, the only timed wait on that path is the dispatcher queue's `poll(keepAlive)`, because `allowCoreThreadTimeOut` is on and the worker stays parked there for the 60 s idle timeout. A worker blocked in the logger can no longer pass, and a worker killed by a rethrow ends `TERMINATED`. The production completion-before-log order is unchanged. **Verification (JDK 17, local)** - `CheckpointCoordinatorTest`, `SharedCheckpointSchedulerTest`, `CheckpointCoordinatorScheduledTriggerTest`, `StateTransitionCleanupTest`: 45 tests, 0 failures. - All `org.apache.seatunnel.engine.server.checkpoint.**` tests plus `SavePointTest` and `CheckpointErrorRestoreEndTest`: 69 run, 0 failures, 2 skipped (existing `@Disabled`). - The two new tests plus the changed scheduler test, 20 consecutive runs: 20/20 green. - `spotless:apply` and `-DskipTests verify` on `seatunnel-engine-server`: clean. Fork Build for `a932e27a7`: https://github.com/Vivek1106-04/seatunnel/actions/runs/40463705801 (running). I will follow up with the result. -- 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]
