code: protected void requestInstances(final ExecutionStage executionStage) throws InstanceException {
  
  		final ExecutionGraph executionGraph = executionStage.getExecutionGraph();
  		final InstanceRequestMap instanceRequestMap = new InstanceRequestMap();
  
  		synchronized (executionStage) {
  
  			executionStage.collectRequiredInstanceTypes(instanceRequestMap, ExecutionState.CREATED);
  
  			final Iterator<Map.Entry<InstanceType, Integer>> it = instanceRequestMap.getMinimumIterator();
  			LOG.info("Requesting the following instances for job " + executionGraph.getJobID());
  			while (it.hasNext()) {
  				final Map.Entry<InstanceType, Integer> entry = it.next();
  				LOG.info(" " + entry.getKey() + " [" + entry.getValue().intValue() + ", "
  					+ instanceRequestMap.getMaximumNumberOfInstances(entry.getKey()) + "]");
  			}
  
  			if (instanceRequestMap.isEmpty()) {
  				return;
  			}
  
  			this.instanceManager.requestInstance(executionGraph.getJobID(), executionGraph.getJobConfiguration(),
  				instanceRequestMap, null);
  
  			// Switch vertex state to assigning
  			final ExecutionGraphIterator it2 = new ExecutionGraphIterator(executionGraph, executionGraph
  				.getIndexOfCurrentExecutionStage(), true, true);
  			while (it2.hasNext()) {
  
  				it2.next().compareAndUpdateExecutionState(ExecutionState.CREATED, ExecutionState.SCHEDULED);
  			}
  		}
  	}
docs: Collects the instances required to run the job from the given {@link ExecutionStage} and requests them at the
  loaded instance manager.
  
  @param executionStage
  the execution stage to collect the required instances from
  @throws InstanceException
  thrown if the given execution graph is already processing its final stage