Repository: apache/flink
File: flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java (Java)

Code before:
     * task has failed, is canceled, or is being canceled at the moment.
     */
    private void releaseResources() {
        LOG.debug(
                "Release task {} network resources (state: {}).",
                taskNameWithSubtask,
                getExecutionState());

        for (ResultPartitionWriter partitionWriter : partitionWriters) {
            taskEventDispatcher.unregisterPartition(partitionWriter.getPartitionId());
        }

        // close network resources
        if (isCanceledOrFailed()) {
            failAllResultPartitions();
        }
        closeAllResultPartitions();
        closeAllInputGates();
        if (this.channelStateWriter != null) {
            LOG.debug("Closing channelStateWriter for task {}", taskNameWithSubtask);
            try {
                this.channelStateWriter.close();
            } catch (IOException e) {
                LOG.error(
                        "Failed to close channelStateWriter for task {}.", taskNameWithSubtask, e);
            }
        }

        try {
            taskStateManager.close();
        } catch (Exception e) {
            LOG.error("Failed to close task state manager for task {}.", taskNameWithSubtask, e);
        }
    }

    private void failAllResultPartitions() {
        for (ResultPartitionWriter partitionWriter : partitionWriters) {
            partitionWriter.fail(getFailureCause());
        }
    }

    private void closeAllResultPartitions() {
        for (ResultPartitionWriter partitionWriter : partitionWriters) {
            try {
                partitionWriter.close();
            } catch (Throwable t) {
                ExceptionUtils.rethrowIfFatalError(t);
                LOG.error(
                        "Failed to release result partition for task {}.", taskNameWithSubtask, t);
            }
        }

Review comment (suggestion):
I am thinking should we catch `Throwable` and call `ExceptionUtils.rethrowIfFatalError` to ensure subsequent closes logic can be executed as expected.

Both of `closeAllResultPartitions` and `closeAllInputGates` have similar catch blocks.

Code after:
     * Releases 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.
     */
    private void releaseResources() {
        LOG.debug(
                "Release task {} network resources (state: {}).",
                taskNameWithSubtask,
                getExecutionState());

        for (ResultPartitionWriter partitionWriter : partitionWriters) {
            taskEventDispatcher.unregisterPartition(partitionWriter.getPartitionId());
        }

        // close network resources
        if (isCanceledOrFailed()) {
            failAllResultPartitions();
        }
        closeAllResultPartitions();
        closeAllInputGates();
        if (this.channelStateWriter != null) {
            LOG.debug("Closing channelStateWriter for task {}", taskNameWithSubtask);
            try {
                this.channelStateWriter.close();
            } catch (Throwable t) {
                ExceptionUtils.rethrowIfFatalError(t);
                LOG.error(
                        "Failed to close channelStateWriter for task {}.", taskNameWithSubtask, t);
            }
        }

        try {
            taskStateManager.close();
        } catch (Exception e) {
            LOG.error("Failed to close task state manager for task {}.", taskNameWithSubtask, e);
        }
    }

    private void failAllResultPartitions() {
        for (ResultPartitionWriter partitionWriter : partitionWriters) {
            partitionWriter.fail(getFailureCause());
        }
    }

    private void closeAllResultPartitions() {
        for (ResultPartitionWriter partitionWriter : partitionWriters) {
            try {
                partitionWriter.close();
            } catch (Throwable t) {
                ExceptionUtils.rethrowIfFatalError(t);
                LOG.error(
                        "Failed to release result partition for task {}.", taskNameWithSubtask, t);