code: private void allVerticesInTerminalState(long expectedGlobalVersionForRestart) {
  
  		assertRunningInJobMasterMainThread();
  
  		// we are done, transition to the final state
  		JobStatus current;
  		while (true) {
  			current = this.state;
  
  			if (current == JobStatus.RUNNING) {
  				failGlobal(new Exception("ExecutionGraph went into allVerticesInTerminalState() from RUNNING"));
  			}
  			else if (current == JobStatus.CANCELLING) {
  				if (transitionState(current, JobStatus.CANCELED)) {
  					onTerminalState(JobStatus.CANCELED);
  					break;
  				}
  			}
  			else if (current == JobStatus.FAILING) {
  				if (tryRestartOrFail(expectedGlobalVersionForRestart)) {
  					break;
  				}
  				// concurrent job status change, let's check again
  			}
  			else if (current.isGloballyTerminalState()) {
  				LOG.warn("Job has entered globally terminal state without waiting for all " +
  					"job vertices to reach final state.");
  				break;
  			}
  			else {
  				failGlobal(new Exception("ExecutionGraph went into final state from state " + current));
  				break;
  			}
  		}
  		// done transitioning the state
  	}
docs: This method is a callback during cancellation/failover and called when all tasks
  have reached a terminal state (cancelled/failed/finished).