code: public void attachJobGraph(List<JobVertex> topologiallySorted) throws JobException {
  
  		assertRunningInJobMasterMainThread();
  
  		LOG.debug("Attaching {} topologically sorted vertices to existing job graph with {} " +
  				"vertices and {} intermediate results.",
  			topologiallySorted.size(),
  			tasks.size(),
  			intermediateResults.size());
  
  		final ArrayList<ExecutionJobVertex> newExecJobVertices = new ArrayList<>(topologiallySorted.size());
  		final long createTimestamp = System.currentTimeMillis();
  
  		for (JobVertex jobVertex : topologiallySorted) {
  
  			if (jobVertex.isInputVertex() && !jobVertex.isStoppable()) {
  				this.isStoppable = false;
  			}
  
  			// create the execution job vertex and attach it to the graph
  			ExecutionJobVertex ejv = new ExecutionJobVertex(
  				this,
  				jobVertex,
  				1,
  				rpcTimeout,
  				globalModVersion,
  				createTimestamp);
  
  			ejv.connectToPredecessors(this.intermediateResults);
  
  			ExecutionJobVertex previousTask = this.tasks.putIfAbsent(jobVertex.getID(), ejv);
  			if (previousTask != null) {
  				throw new JobException(String.format("Encountered two job vertices with ID %s : previous=[%s] / new=[%s]",
  					jobVertex.getID(), ejv, previousTask));
  			}
  
  			for (IntermediateResult res : ejv.getProducedDataSets()) {
  				IntermediateResult previousDataSet = this.intermediateResults.putIfAbsent(res.getId(), res);
  				if (previousDataSet != null) {
  					throw new JobException(String.format("Encountered two intermediate data set with ID %s : previous=[%s] / new=[%s]",
  						res.getId(), res, previousDataSet));
  				}
  			}
  
  			this.verticesInCreationOrder.add(ejv);
  			this.numVerticesTotal += ejv.getParallelism();
  			newExecJobVertices.add(ejv);
  		}
  
  		terminationFuture = new CompletableFuture<>();
  		failoverStrategy.notifyNewVertices(newExecJobVertices);
  	}
docs: --------------------------------------------------------------------------------------------