XComp commented on a change in pull request #14847:
URL: https://github.com/apache/flink/pull/14847#discussion_r578568213
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerBase.java
##########
@@ -908,49 +909,38 @@ public void reportCheckpointMetrics(
// will be restarted by the CheckpointCoordinatorDeActivator.
checkpointCoordinator.stopCheckpointScheduler();
+ final CompletableFuture<Collection<ExecutionState>>
executionTerminationsFuture =
+ getCombinedExecutionTerminationFuture();
+
final CompletableFuture<String> savepointFuture =
checkpointCoordinator
.triggerSynchronousSavepoint(advanceToEndOfEventTime,
targetDirectory)
.thenApply(CompletedCheckpoint::getExternalPointer);
- final CompletableFuture<JobStatus> terminationFuture =
- executionGraph
- .getTerminationFuture()
- .handle(
- (jobstatus, throwable) -> {
- if (throwable != null) {
- log.info(
- "Failed during stopping job {}
with a savepoint. Reason: {}",
- jobGraph.getJobID(),
- throwable.getMessage());
- throw new
CompletionException(throwable);
- } else if (jobstatus !=
JobStatus.FINISHED) {
- log.info(
- "Failed during stopping job {}
with a savepoint. Reason: Reached state {} instead of FINISHED.",
- jobGraph.getJobID(),
- jobstatus);
- throw new CompletionException(
- new FlinkException(
- "Reached state "
- + jobstatus
- + " instead of
FINISHED."));
- }
- return jobstatus;
- });
-
- return savepointFuture
- .thenCompose((path) -> terminationFuture.thenApply((jobStatus
-> path)))
- .handleAsync(
- (path, throwable) -> {
- if (throwable != null) {
- // restart the checkpoint coordinator if
stopWithSavepoint failed.
-
startCheckpointScheduler(checkpointCoordinator);
- throw new CompletionException(throwable);
- }
+ StopWithSavepointContext stopWithSavepointContext =
Review comment:
I renamed the class into `StopWithSavepointOperationsImpl`.
----------------------------------------------------------------
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:
us...@infra.apache.org
