code: public void registerJob(final ExecutionGraph executionGraph, final boolean profilingAvailable,
  			final long submissionTimestamp) {
  
  		final Iterator<ExecutionVertex> it = new ExecutionGraphIterator(executionGraph, true);
  
  		while (it.hasNext()) {
  
  			final ExecutionVertex vertex = it.next();
  
  			// Register the listener object which will pass state changes on to the collector
  			vertex.registerExecutionListener(new ExecutionListenerWrapper(this, vertex));
  
  			// Register the listener object which will pass assignment changes on to the collector
  			vertex.registerVertexAssignmentListener(new VertexAssignmentListenerWrapper(this, executionGraph.getJobID()));
  		}
  
  		// Register one job status listener wrapper for the entire job
  		executionGraph.registerJobStatusListener(new JobStatusListenerWrapper(this, executionGraph.getJobName(),
  			profilingAvailable, submissionTimestamp));
  
  	}
docs: Registers a job in form of its execution graph representation
  with the job progress collector. The collector will subscribe
  to state changes of the individual subtasks. A separate
  deregistration is not necessary since the job progress collector
  periodically discards outdated progress information.
  
  @param executionGraph
  the execution graph representing the job
  @param profilingAvailable
  indicates if profiling data is available for this job
  @param submissionTimestamp
  the submission time stamp of the job