Vivek1106-04 opened a new issue, #13052:
URL: https://github.com/apache/seatunnel/issues/13052
### What happened
Several checkpoint error handlers in `CheckpointCoordinator` run later, on
the coordinator executor or on the checkpoint timer thread, and report failures
through `handleCoordinatorError`. That method only checks the coordinator
future that is current *when the handler runs*
(`CheckpointCoordinator.java:354-371` on `dev`):
```java
private void handleCoordinatorError(CheckpointCloseReason reason, Throwable
e) {
CheckpointException checkpointException = new
CheckpointException(reason, e);
errorByPhysicalVertex.compareAndSet(null,
ExceptionUtils.getMessage(checkpointException));
if (checkpointCoordinatorFuture.isDone()) {
return;
}
updateStatus(CheckpointCoordinatorStatus.FAILED);
checkpointCoordinatorFuture.complete(...FAILED...);
checkpointManager.handleCheckpointError(pipelineId, false);
cleanPendingCheckpoint(reason);
}
```
`restoreCoordinator` (`CheckpointCoordinator.java:704`) reuses the same
coordinator for a pipeline restore. It replaces `errorByPhysicalVertex` and
`checkpointCoordinatorFuture` (line 707), sets the status to `RUNNING`, and
cleans the old pending checkpoints. Nothing ties an in-flight handler to the
generation that started its work. So a handler of the previous generation that
runs after the swap:
- fails the restored generation: its future completes `FAILED`,
`handleCheckpointError(pipelineId, false)` is called again, and its new pending
state is cleaned;
- writes the old error into the restored generation's
`errorByPhysicalVertex`, even when the restored future is already done.
The deferred handlers, on `dev`:
| Handler message | Location | Runs on |
| --- | --- | --- |
| `trigger checkpoint failed` | `:982` | coordinator executor
(`whenCompleteAsync`) |
| `complete checkpoint failed` | `:990` | coordinator executor |
| `triggering checkpoint barrier has been interrupted` / `... failed` |
`:1020`, `:1026` | coordinator executor, after `allOf(...).get()` on the
barrier RPCs |
| `get checkpoint id failed` | `:1080` | coordinator executor
(`supplyAsync`) |
| periodic trigger wrapper (`Failed to trigger ...`) | `:557` | checkpoint
timer thread |
| checkpoint timeout | `:1054` | checkpoint timer thread |
The cleanup in the restore cannot stop these handlers. The barrier wait and
the completion callback are not scheduled tasks. The cleanup only cancels timer
tasks that have not started yet, so a trigger body that is already running
finishes and reports its error.
### How it shows up
A member leaves while a checkpoint barrier is in flight. The pipeline fails
and goes through its restore, and `restoreCoordinator` starts a new generation.
The barrier RPC to the lost member then fails, `allOf(...).get()` throws, and
the old trigger body reports `triggering checkpoint barrier failed`. The
restored pipeline is failed again, with an error that belongs to the attempt it
already restored from. Each extra failure uses up one of the pipeline's restore
attempts, and the job can end `FAILED` even though the restored attempt was
healthy.
This is a narrow race, and I have not seen it reported from production. It
was found in review of #12165 ([review
comment](https://github.com/apache/seatunnel/pull/12165#pullrequestreview-5667062792),
Issue 1), and a controlled-executor test reproduces it deterministically (see
below).
### Expected behavior
An asynchronous error fails only the coordinator generation that started the
failed work. If that generation has already ended, the error is logged and
dropped. Synchronous callers (`reportCheckpointErrorFromTask`,
`notifyCompleted` on the restore path) run in the generation that raised the
error and are unaffected.
### Context
- #12165 adds this fence for the checkpoint timeout only: `expireCheckpoint`
carries the generation's future, and `failExpiredCheckpoint` re-checks it under
a `generationLock` that `restoreCoordinator` also holds for the swap. This
issue covers the remaining deferred handlers listed above.
- Planned fix: the deferred handlers capture the current future when their
work starts and call an owner-checked `handleCoordinatorError(owner, ...)` that
uses the same `generationLock` check, with the timeout path sharing the same
fail step. The fix is ready locally on top of #12165, and I will open the PR
once #12165 merges. Regression tests use held executors: the old completion
error or trigger error, released after a restore, must leave the restored
generation running, and the same error without a restore must still fail it.
With the owner removed from a call site, the corresponding test fails.
- The same PR will also make `expireCheckpoint` package-private with
`@VisibleForTesting`, so tests do not call it by reflection, and add a test
that an expiry during `startSavepoint`'s lock-held poll fails the coordinator
and ends the poll (Issues 2 and 3 in the same review).
--
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]
