mohsenrezaeithe commented on code in PR #28201:
URL: https://github.com/apache/flink/pull/28201#discussion_r4472896445
##########
flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterServiceLeadershipRunner.java:
##########
@@ -137,13 +152,12 @@ public CompletableFuture<Void> closeAsync() {
new FlinkException(
"JobMasterServiceLeadershipRunner is closed.
Therefore, the corresponding JobMaster will never acquire the leadership."));
- resultFuture.complete(
- JobManagerRunnerResult.forSuccess(
-
createExecutionGraphInfoWithJobStatus(JobStatus.SUSPENDED)));
-
processTerminationFuture = jobMasterServiceProcess.closeAsync();
}
+ processTerminationFuture.whenComplete(
+ (ignored, throwable) -> completeResultFutureAfterClose());
Review Comment:
Previously the runner's result was completed with `SUSPENDED` right away.
Now it completes only after `jobMasterServiceProcess.closeAsync()` finishes,
which has no upper bound. A slow JobMaster stop (checkpoint-store shutdown I/O,
slow task cancellation) would delay the Dispatcher's handling of this runner
during shutdown and leadership changes.
Is that intended? If so, a short comment here, and maybe a test with a
process termination that never completes, would make the behaviour explicit.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/DefaultJobMasterServiceProcess.java:
##########
@@ -154,18 +153,32 @@ public CompletableFuture<Void> closeAsync() {
jobId,
leaderSessionId);
- resultFuture.completeExceptionally(new
JobNotFinishedException(jobId));
jobMasterGatewayFuture.completeExceptionally(
new FlinkException("Process has been closed."));
jobMasterServiceFuture.whenComplete(
(jobMasterService, throwable) -> {
if (throwable != null) {
- // JobMasterService creation has failed.
Nothing to stop then :-)
+ resultFuture.completeExceptionally(
+ new JobNotFinishedException(jobId));
terminationFuture.complete(null);
} else {
- FutureUtils.forward(
- jobMasterService.closeAsync(),
terminationFuture);
+ // Defer JobNotFinishedException until after
the JobMasterService
+ // has fully closed, so any in-flight terminal
completion from the
+ // scheduler (jobReachedGloballyTerminalState)
can win the race and
+ // be observed by the runner. See FLINK-39704.
+ jobMasterService
+ .closeAsync()
+ .whenComplete(
+ (unused, t) -> {
+
resultFuture.completeExceptionally(
+ new
JobNotFinishedException(jobId));
+ if (t != null) {
+
terminationFuture.completeExceptionally(t);
+ } else {
+
terminationFuture.complete(null);
+ }
+ });
Review Comment:
Deferring the `JobNotFinishedException` until
`jobMasterService.closeAsync()` settles makes sense. While testing on a 2.3
job, though, I noticed the terminal result can still arrive after this point.
The status reaches this class through two asynchronous hops:
- `JobManagerJobStatusListener#jobStatusChanges` → `runAsync(() ->
jobStatusChanged(...))`
- then `futureExecutor.execute(...)` in `JobMaster#jobStatusChanged`, after
`partitionPromoteFuture.get()`
Neither hop is ordered before `JobMaster#onStop`, and `stopScheduling()`
stops the status listener. If the stop wins, `resultFuture` is completed here
with `JobNotFinishedException`, the later `complete(FAILED)` is a no-op, and
the runner never caches the terminal result, while the scheduler still shuts
its checkpoint services down with `FAILED`.
One option I tried locally: in `JobMaster#stopScheduling()`, before
`schedulerNG.closeAsync()`, publish the scheduler's own state if it is already
globally terminal:
```java
if (schedulerNG.requestJobStatus().isGloballyTerminalState()) {
jobCompletionActions.jobReachedGloballyTerminalState(schedulerNG.requestJob());
}
```
This runs on the main thread inside `onStop`, so together with this deferral
the terminal result is always published first. Publishing it twice looks
harmless, since this class completes a first-wins future. A `JobMasterTest`
case that stops the JobMaster while the scheduler is FAILED, CANCELED or
FINISHED (plus a non-terminal control) passes with it and fails without it, and
the existing tests and the ones in this PR still pass.
Happy to share the test, if you'd like to add this fix in this PR as well.
--
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]
