code: private Task createAndRegisterTask(final ExecutionVertexID id, final Configuration jobConfiguration,
  			final RuntimeEnvironment environment)
  					throws InsufficientResourcesException, IOException {
  
  		if (id == null) {
  			throw new IllegalArgumentException("Argument id is null");
  		}
  
  		if (environment == null) {
  			throw new IllegalArgumentException("Argument environment is null");
  		}
  
  		// Task creation and registration must be atomic
  		Task task;
  
  		synchronized (this) {
  			final Task runningTask = this.runningTasks.get(id);
  			boolean registerTask = true;
  			if (runningTask == null) {
  				task = new Task(id, environment, this);
  			} else {
  
  				if (runningTask instanceof Task) {
  					// Task is already running
  					return null;
  				} else {
  					// There is already a replay task running, we will simply restart it
  					task = runningTask;
  					registerTask = false;
  				}
  
  			}
  
  			if (registerTask) {
  				// Register the task with the byte buffered channel manager
  				this.channelManager.register(task);
  
  				boolean enableProfiling = false;
  				if (this.profiler != null && jobConfiguration.getBoolean(ProfilingUtils.PROFILE_JOB_KEY, true)) {
  					enableProfiling = true;
  				}
  
  				// Register environment, input, and output gates for profiling
  				if (enableProfiling) {
  					task.registerProfiler(this.profiler, jobConfiguration);
  				}
  
  				this.runningTasks.put(id, task);
  			}
  		}
  		return task;
  	}
docs: Registers an newly incoming runtime task with the task manager.
  
  @param id
  the ID of the task to register
  @param jobConfiguration
  the job configuration that has been attached to the original job graph
  @param environment
  the environment of the task to be registered
  @return the task to be started or <code>null</code> if a task with the same ID was already running