tillrohrmann commented on a change in pull request #7568: [FLINK-11417] Make
access to ExecutionGraph single threaded from JobMaster main thread
URL: https://github.com/apache/flink/pull/7568#discussion_r253421327
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/FailoverRegion.java
##########
@@ -138,34 +136,36 @@ else if (curStatus.equals(JobStatus.CANCELED)) {
// cancel all executions in this sub graph
private void cancel(final long globalModVersionOfFailover) {
+
executionGraph.getJobMasterMainThreadExecutor().assertRunningInMainThread();
while (true) {
JobStatus curStatus = this.state;
if (curStatus.equals(JobStatus.RUNNING)) {
if (transitionState(curStatus,
JobStatus.CANCELLING)) {
- // we build a future that is complete
once all vertices have reached a terminal state
- final ArrayList<CompletableFuture<?>>
futures = new ArrayList<>(connectedExecutionVertexes.size());
-
- // cancel all tasks (that still need
cancelling)
- for (ExecutionVertex vertex :
connectedExecutionVertexes) {
- futures.add(vertex.cancel());
- }
-
- final FutureUtils.ConjunctFuture<Void>
allTerminal = FutureUtils.waitForAll(futures);
- allTerminal.thenAcceptAsync(
- (Void value) ->
allVerticesInTerminalState(globalModVersionOfFailover),
- executor);
-
+
createTerminationFutureOverAllConnectedVertexes()
+ .thenAccept((nullptr) ->
allVerticesInTerminalState(globalModVersionOfFailover));
break;
}
- }
- else {
+ } else {
LOG.info("FailoverRegion {} is {} when
cancel.", id, state);
break;
}
}
}
+ @VisibleForTesting
+ protected CompletableFuture<Void>
createTerminationFutureOverAllConnectedVertexes() {
Review comment:
Good refactoring.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on 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
