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

Code before:
            }

            // 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)) {
                    break;
                }
            } else if (current == ExecutionState.FAILED) {
                // in state failed already, no transition necessary any more
                break;
            }

Review comment (performance):
I suggest adding a `yield()` call in the loop to prevent this thread hogging the cpu in a tight loop until done.  

Code after:
            }

            // 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)) {
                    break;
                }
            } else if (current == ExecutionState.FAILED) {
                // in state failed already, no transition necessary any more
                break;
            }