code: @Override
  	public CompletableFuture<Acknowledge> updatePartitions(
  			final ExecutionAttemptID executionAttemptID,
  			Iterable<PartitionInfo> partitionInfos,
  			Time timeout) {
  		final Task task = taskSlotTable.getTask(executionAttemptID);
  
  		if (task != null) {
  			for (final PartitionInfo partitionInfo: partitionInfos) {
  				IntermediateDataSetID intermediateResultPartitionID = partitionInfo.getIntermediateDataSetID();
  
  				final SingleInputGate singleInputGate = task.getInputGateById(intermediateResultPartitionID);
  
  				if (singleInputGate != null) {
  					// Run asynchronously because it might be blocking
  					getRpcService().execute(
  						() -> {
  							try {
  								singleInputGate.updateInputChannel(partitionInfo.getInputChannelDeploymentDescriptor());
  							} catch (IOException | InterruptedException e) {
  								log.error("Could not update input data location for task {}. Trying to fail task.", task.getTaskInfo().getTaskName(), e);
  
  								try {
  									task.failExternally(e);
  								} catch (RuntimeException re) {
  									// TODO: Check whether we need this or make exception in failExtenally checked
  									log.error("Failed canceling task with execution ID {} after task update failure.", executionAttemptID, re);
  								}
  							}
  						});
  				} else {
  					return FutureUtils.completedExceptionally(
  						new PartitionException("No reader with ID " + intermediateResultPartitionID +
  							" for task " + executionAttemptID + " was found."));
  				}
  			}
  
  			return CompletableFuture.completedFuture(Acknowledge.get());
  		} else {
  			log.debug("Discard update for input partitions of task {}. Task is no longer running.", executionAttemptID);
  			return CompletableFuture.completedFuture(Acknowledge.get());
  		}
  	}
docs: ----------------------------------------------------------------------