code: @Override
  	public void executionStateChanged(final ExecutionState newExecutionState, final String optionalMessage) {
  
  		// Check the state transition
  		ExecutionStateTransition.checkTransition(false, getTaskName(), this.executionState, newExecutionState);
  
  		// Make sure the reason for a transition to FAILED appears in the log files
  		if (newExecutionState == ExecutionState.FAILED) {
  			LOG.error(optionalMessage);
  		}
  
  		// Notify all listener objects
  		final Iterator<ExecutionListener> it = this.registeredListeners.iterator();
  		while (it.hasNext()) {
  			it.next().executionStateChanged(this.environment.getJobID(), this.vertexID, newExecutionState,
  					optionalMessage);
  		}
  
  		// Store the new execution state
  		this.executionState = newExecutionState;
  
  		// Finally propagate the state change to the job manager
  		this.taskManager.executionStateChanged(this.environment.getJobID(), this.vertexID, newExecutionState,
  				optionalMessage);
  	}
docs: -----------------------------------------------------------------------------------------------------------------