tillrohrmann commented on a change in pull request #7889: [FLINK-11665] Wait
for job termination before recovery in Dispatcher
URL: https://github.com/apache/flink/pull/7889#discussion_r262440270
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
##########
@@ -897,27 +923,17 @@ public void grantLeadership(final UUID
newLeaderSessionID) {
}
}
- private CompletableFuture<Void> waitForTerminatingJobManager(JobID
jobId, JobGraph jobGraph, FunctionWithException<JobGraph,
CompletableFuture<Void>, ?> action) {
- final CompletableFuture<Void> jobManagerTerminationFuture =
getJobTerminationFuture(jobId)
- .exceptionally((Throwable throwable) -> {
- throw new CompletionException(
- new DispatcherException(
- String.format("Termination of
previous JobManager for job %s failed. Cannot submit job under the same job
id.", jobId),
- throwable)); });
-
- return jobManagerTerminationFuture.thenComposeAsync(
- FunctionUtils.uncheckedFunction((ignored) -> {
- jobManagerTerminationFutures.remove(jobId);
- return action.apply(jobGraph);
- }),
- getMainThreadExecutor());
- }
-
- CompletableFuture<Void> getJobTerminationFuture(JobID jobId) {
+ CompletableFuture<JobID> getJobTerminationFuture(JobID jobId) {
if (jobManagerRunnerFutures.containsKey(jobId)) {
return FutureUtils.completedExceptionally(new
DispatcherException(String.format("Job with job id %s is still running.",
jobId)));
} else {
- return jobManagerTerminationFutures.getOrDefault(jobId,
CompletableFuture.completedFuture(null));
+ return jobManagerTerminationFutures.getOrDefault(jobId,
CompletableFuture.completedFuture(null))
+ .exceptionally((Throwable throwable) -> {
+ throw new CompletionException(
+ new DispatcherException(
+
String.format("Termination of previous JobManager for job %s failed. Cannot
submit job under the same job id.", jobId),
+ throwable)); })
+ .thenApply(ignored -> jobId);
Review comment:
the `thenApply` can be removed if `CompletableFuture.completedFuture(jobId)`
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
