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

Code before:
                String message =
                        String.format(
                                "FATAL - exception in resource cleanup of task %s (%s).",
                                taskNameWithSubtask, executionId);
                LOG.error(message, t);
                notifyFatalError(message, t);
            }

            // un-register the metrics at the end so that the task may already be
            // counted as finished when this happens
            // errors here will only be logged
            try {
                metrics.close();
            } catch (Throwable t) {
                LOG.error(
                        "Error during metrics de-registration of task {} ({}).",
                        taskNameWithSubtask,
                        executionId,
                        t);
            }
        }
    }

    /**
     * Transition into our final state in case of failure. We should be either in DEPLOYING,
     * INITIALIZING, RUNNING, CANCELING, or FAILED loop for multiple retries during concurrent state
     * changes via calls to cancel() or to failExternally()
     */
    private void transitionStateOnFailure(
            Throwable t, AutoCloseableRegistry postFailureCleanUpRegistry) throws IOException {
        while (true) {
            ExecutionState current = this.executionState;

            if (current == ExecutionState.RUNNING
                    || current == ExecutionState.INITIALIZING
                    || current == ExecutionState.DEPLOYING) {
                if (ExceptionUtils.findThrowable(t, CancelTaskException.class).isPresent()) {
                    if (transitionState(current, ExecutionState.CANCELED, t)) {
                        postFailureCleanUpRegistry.registerCloseable(
                                () -> cancelInvokable(invokable));
                        break;
                    }
                } else {
                    if (transitionState(current, ExecutionState.FAILED, t)) {
                        postFailureCleanUpRegistry.registerCloseable(
                                () -> cancelInvokable(invokable));
                        break;
                    }
                }
            } else if (current == ExecutionState.CANCELING) {
                if (transitionState(current, ExecutionState.CANCELED)) {

Review comment (bug):
nit: The sentence does not read well. I suggest 
- full stop after FAILED
- then `Loop to asynchronously clean up via calls to cancel() or to failExternally()`

Code after:
                String message =
                        String.format(
                                "FATAL - exception in resource cleanup of task %s (%s).",
                                taskNameWithSubtask, executionId);
                LOG.error(message, t);
                notifyFatalError(message, t);
            }

            // un-register the metrics at the end so that the task may already be
            // counted as finished when this happens
            // errors here will only be logged
            try {
                metrics.close();
            } catch (Throwable t) {
                LOG.error(
                        "Error during metrics de-registration of task {} ({}).",
                        taskNameWithSubtask,
                        executionId,
                        t);
            }
        }
    }

    /**
     * Transition into our final state in case of failure. We should be either in DEPLOYING,
     * INITIALIZING, RUNNING, CANCELING, or FAILED. Loop for multiple retries in case of concurrent
     * state changes via calls to cancel() or to failExternally()
     */
    private void transitionStateOnFailure(
            Throwable t, AutoCloseableRegistry postFailureCleanUpRegistry) throws IOException {
        while (true) {
            ExecutionState current = this.executionState;

            if (current == ExecutionState.RUNNING
                    || current == ExecutionState.INITIALIZING
                    || current == ExecutionState.DEPLOYING) {
                if (ExceptionUtils.findThrowable(t, CancelTaskException.class).isPresent()) {
                    if (transitionState(current, ExecutionState.CANCELED, t)) {
                        postFailureCleanUpRegistry.registerCloseable(
                                () -> cancelInvokable(invokable));
                        break;
                    }
                } else {
                    if (transitionState(current, ExecutionState.FAILED, t)) {
                        postFailureCleanUpRegistry.registerCloseable(
                                () -> cancelInvokable(invokable));
                        break;
                    }
                }
            } else if (current == ExecutionState.CANCELING) {
                if (transitionState(current, ExecutionState.CANCELED)) {