code: public void deployAssignedInputVertices(final ExecutionGraph executionGraph) {
  
  		final Map<AbstractInstance, List<ExecutionVertex>> verticesToBeDeployed = new HashMap<AbstractInstance, List<ExecutionVertex>>();
  		final ExecutionStage executionStage = executionGraph.getCurrentExecutionStage();
  
  		final Set<ExecutionVertex> alreadyVisited = new HashSet<ExecutionVertex>();
  
  		for (int i = 0; i < executionStage.getNumberOfStageMembers(); ++i) {
  
  			final ExecutionGroupVertex startVertex = executionStage.getStageMember(i);
  			if (!startVertex.isInputVertex()) {
  				continue;
  			}
  
  			for (int j = 0; j < startVertex.getCurrentNumberOfGroupMembers(); ++j) {
  				final ExecutionVertex vertex = startVertex.getGroupMember(j);
  				findVerticesToBeDeployed(vertex, verticesToBeDeployed, alreadyVisited);
  			}
  		}
  
  		if (!verticesToBeDeployed.isEmpty()) {
  
  			final Iterator<Map.Entry<AbstractInstance, List<ExecutionVertex>>> it2 = verticesToBeDeployed
  				.entrySet()
  				.iterator();
  
  			while (it2.hasNext()) {
  
  				final Map.Entry<AbstractInstance, List<ExecutionVertex>> entry = it2.next();
  				this.deploymentManager.deploy(executionGraph.getJobID(), entry.getKey(), entry.getValue());
  			}
  		}
  	}
docs: Collects all execution vertices with the state ASSIGNED starting from the input vertices of the current execution
  stage and deploys them on the assigned {@link AllocatedResource} objects.
  
  @param executionGraph
  the execution graph to collect the vertices from