code: public void registerJob(final ExecutionGraph executionGraph) {
  
  		final Iterator<ExecutionGroupVertex> it = new ExecutionGroupVertexIterator(executionGraph, true, -1);
  		while (it.hasNext()) {
  
  			final ExecutionGroupVertex groupVertex = it.next();
  			final InputSplit[] inputSplits = groupVertex.getInputSplits();
  
  			if (inputSplits == null) {
  				continue;
  			}
  
  			if (inputSplits.length == 0) {
  				continue;
  			}
  
  			final AbstractInvokable invokable = groupVertex.getEnvironment().getInvokable();
  			if (!(invokable instanceof AbstractInputTask)) {
  				LOG.error(groupVertex.getName() + " has " + inputSplits.length
  					+ " input splits, but is not of typt AbstractInputTask, ignoring...");
  				continue;
  			}
  
  			@SuppressWarnings("unchecked")
  			final AbstractInputTask<? extends InputSplit> inputTask = (AbstractInputTask<? extends InputSplit>) invokable;
  			final Class<? extends InputSplit> splitType = inputTask.getInputSplitType();
  
  			final InputSplitAssigner assigner = getAssignerByType(splitType, true);
  			// Add entry to cache for fast retrieval during the job execution
  			this.assignerCache.put(groupVertex, assigner);
  
  			assigner.registerGroupVertex(groupVertex);
  		}
  
  		// Register job with the input split tracker
  		this.inputSplitTracker.registerJob(executionGraph);
  	}
docs: Registers a new job represented by its {@link ExecutionGraph} with the input split manager.
  
  @param executionGraph
  the job to be registered