code: private void releaseNetworkResources() {
  		LOG.debug("Release task {} network resources (state: {}).", taskNameWithSubtask, getExecutionState());
  
  		for (ResultPartition partition : producedPartitions) {
  			taskEventDispatcher.unregisterPartition(partition.getPartitionId());
  			if (isCanceledOrFailed()) {
  				partition.fail(getFailureCause());
  			}
  		}
  
  		closeNetworkResources();
  	}
docs: Releases network resources before task exits. We should also fail the partition to release if the task
  has failed, is canceled, or is being canceled at the moment.