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 ( ) ; }
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 .