code: private void restart(long globalModVersionOfFailover) {
  		try {
  			if (transitionState(JobStatus.CREATED, JobStatus.RUNNING)) {
  				// if we have checkpointed state, reload it into the executions
  				if (executionGraph.getCheckpointCoordinator() != null) {
  					// we restart the checkpoint scheduler for
  					// i) enable new checkpoint could be triggered without waiting for last checkpoint expired.
  					// ii) ensure the EXACTLY_ONCE semantics if needed.
  					executionGraph.getCheckpointCoordinator().abortPendingCheckpoints(
  						new CheckpointException(CheckpointFailureReason.JOB_FAILOVER_REGION));
  
  					executionGraph.getCheckpointCoordinator().restoreLatestCheckpointedState(
  						tasks, false, true);
  				}
  
  				HashSet<AllocationID> previousAllocationsInRegion = new HashSet<>(connectedExecutionVertexes.size());
  				for (ExecutionVertex connectedExecutionVertex : connectedExecutionVertexes) {
  					AllocationID latestPriorAllocation = connectedExecutionVertex.getLatestPriorAllocation();
  					if (latestPriorAllocation != null) {
  						previousAllocationsInRegion.add(latestPriorAllocation);
  					}
  				}
  
  				//TODO, use restart strategy to schedule them.
  				//restart all connected ExecutionVertexes
  				for (ExecutionVertex ev : connectedExecutionVertexes) {
  					try {
  						ev.scheduleForExecution(
  							executionGraph.getSlotProvider(),
  							executionGraph.isQueuedSchedulingAllowed(),
  							LocationPreferenceConstraint.ANY,
  							previousAllocationsInRegion); // some inputs not belonging to the failover region might have failed concurrently
  					}
  					catch (Throwable e) {
  						failover(globalModVersionOfFailover);
  					}
  				}
  			}
  			else {
  				LOG.info("FailoverRegion {} switched from CREATED to RUNNING fail, will fail this region again.", id);
  				failover(globalModVersionOfFailover);
  			}
  		} catch (Exception e) {
  			LOG.info("FailoverRegion {} restart failed, failover again.", id, e);
  			failover(globalModVersionOfFailover);
  		}
  	}
docs: restart all executions in this sub graph