docs: @Override
  	public void triggerPartitionProducerStateCheck(
  		JobID jobId,
  		final IntermediateDataSetID intermediateDataSetId,
  		final ResultPartitionID resultPartitionId) {
  
  		CompletableFuture<ExecutionState> futurePartitionState =
  			partitionProducerStateChecker.requestPartitionProducerState(
  				jobId,
  				intermediateDataSetId,
  				resultPartitionId);
  
  		futurePartitionState.whenCompleteAsync(
  			(ExecutionState executionState, Throwable throwable) -> {
  				try {
  					if (executionState != null) {
  						onPartitionStateUpdate(
  							intermediateDataSetId,
  							resultPartitionId,
  							executionState);
  					} else if (throwable instanceof TimeoutException) {
  						// our request timed out, assume we're still running and try again
  						onPartitionStateUpdate(
  							intermediateDataSetId,
  							resultPartitionId,
  							ExecutionState.RUNNING);
  					} else if (throwable instanceof PartitionProducerDisposedException) {
  						String msg = String.format("Producer %s of partition %s disposed. Cancelling execution.",
  							resultPartitionId.getProducerId(), resultPartitionId.getPartitionId());
  						LOG.info(msg, throwable);
  						cancelExecution();
  					} else {
  						failExternally(throwable);
  					}
  				} catch (IOException | InterruptedException e) {
  					failExternally(e);
  				}
  			},
  			executor);
  	}