docs: public void failGlobal(Throwable t) {
  
  		assertRunningInJobMasterMainThread();
  
  		while (true) {
  			JobStatus current = state;
  			// stay in these states
  			if (current == JobStatus.FAILING ||
  				current == JobStatus.SUSPENDED ||
  				current.isGloballyTerminalState()) {
  				return;
  			} else if (transitionState(current, JobStatus.FAILING, t)) {
  				initFailureCause(t);
  
  				// make sure no concurrent local or global actions interfere with the failover
  				final long globalVersionForRestart = incrementGlobalModVersion();
  
  				final CompletableFuture<Void> ongoingSchedulingFuture = schedulingFuture;
  
  				// cancel ongoing scheduling action
  				if (ongoingSchedulingFuture != null) {
  					ongoingSchedulingFuture.cancel(false);
  				}
  
  				// we build a future that is complete once all vertices have reached a terminal state
  				final ArrayList<CompletableFuture<?>> futures = new ArrayList<>(verticesInCreationOrder.size());
  
  				// cancel all tasks (that still need cancelling)
  				for (ExecutionJobVertex ejv : verticesInCreationOrder) {
  					futures.add(ejv.cancelWithFuture());
  				}
  
  				final ConjunctFuture<Void> allTerminal = FutureUtils.waitForAll(futures);
  				allTerminal.whenComplete(
  					(Void ignored, Throwable throwable) -> {
  						if (throwable != null) {
  							transitionState(
  								JobStatus.FAILING,
  								JobStatus.FAILED,
  								new FlinkException("Could not cancel all execution job vertices properly.", throwable));
  						} else {
  							allVerticesInTerminalState(globalVersionForRestart);
  						}
  					});
  
  				return;
  			}
  
  			// else: concurrent change to execution state, retry
  		}
  	}