code: @VisibleForTesting
  	void onPartitionStateUpdate(
  			IntermediateDataSetID intermediateDataSetId,
  			ResultPartitionID resultPartitionId,
  			ExecutionState producerState) throws IOException, InterruptedException {
  
  		if (executionState == ExecutionState.RUNNING) {
  			final SingleInputGate inputGate = inputGatesById.get(intermediateDataSetId);
  
  			if (inputGate != null) {
  				if (producerState == ExecutionState.SCHEDULED
  					|| producerState == ExecutionState.DEPLOYING
  					|| producerState == ExecutionState.RUNNING
  					|| producerState == ExecutionState.FINISHED) {
  
  					// Retrigger the partition request
  					inputGate.retriggerPartitionRequest(resultPartitionId.getPartitionId());
  
  				} else if (producerState == ExecutionState.CANCELING
  					|| producerState == ExecutionState.CANCELED
  					|| producerState == ExecutionState.FAILED) {
  
  					// The producing execution has been canceled or failed. We
  					// don't need to re-trigger the request since it cannot
  					// succeed.
  					if (LOG.isDebugEnabled()) {
  						LOG.debug("Cancelling task {} after the producer of partition {} with attempt ID {} has entered state {}.",
  							taskNameWithSubtask,
  							resultPartitionId.getPartitionId(),
  							resultPartitionId.getProducerId(),
  							producerState);
  					}
  
  					cancelExecution();
  				} else {
  					// Any other execution state is unexpected. Currently, only
  					// state CREATED is left out of the checked states. If we
  					// see a producer in this state, something went wrong with
  					// scheduling in topological order.
  					String msg = String.format("Producer with attempt ID %s of partition %s in unexpected state %s.",
  						resultPartitionId.getProducerId(),
  						resultPartitionId.getPartitionId(),
  						producerState);
  
  					failExternally(new IllegalStateException(msg));
  				}
  			} else {
  				failExternally(new IllegalStateException("Received partition producer state for " +
  						"unknown input gate " + intermediateDataSetId + "."));
  			}
  		} else {
  			LOG.debug("Task {} ignored a partition producer state notification, because it's not running.", taskNameWithSubtask);
  		}
  	}
docs: Answer to a partition state check issued after a failed partition request.