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

Code before:
                }
                return;
            } else if (current == ExecutionState.CANCELING) {
                if (transitionState(ExecutionState.CANCELING, ExecutionState.CANCELED)) {
                    // we were immediately canceled. tell the TaskManager that we reached our final
                    // state
                    notifyFinalState();
                    if (metrics != null) {
                        metrics.close();
                    }
                    return;
                }
            } else {
                if (metrics != null) {
                    metrics.close();
                }
                throw new IllegalStateException(
                        "Invalid state for beginning of operation of task " + this + '.');
            }
        }

        // all resource acquisitions and registrations from here on
        // need to be undone in the end
        Map<String, Future<Path>> distributedCacheEntries = new HashMap<>();
        TaskInvokable invokable = null;
        AutoCloseableRegistry cleanUpRegistry = new AutoCloseableRegistry();

        try {
            // ----------------------------
            //  Task Bootstrap - We periodically
            //  check for canceling as a shortcut
            // ----------------------------

            // activate safety net for task thread
            LOG.debug("Creating FileSystem stream leak safety net for task {}", this);
            FileSystemSafetyNet.initializeSafetyNetForThread();

            // first of all, get a user-code classloader
            // this may involve downloading the job's JAR files and/or classes
            LOG.info("Loading JAR files for task {}.", this);

            userCodeClassLoader = createUserCodeClassloader();
            final ExecutionConfig executionConfig =
                    serializedExecutionConfig.deserializeValue(userCodeClassLoader.asClassLoader());
            Configuration executionConfigConfiguration = executionConfig.toConfiguration();

            // override task cancellation interval from Flink config if set in ExecutionConfig
            taskCancellationInterval =
                    executionConfigConfiguration
                            .getOptional(TaskManagerOptions.TASK_CANCELLATION_INTERVAL)
                            .orElse(Duration.ofMillis(taskCancellationInterval))

Review comment (bug):
I notice the registry is created for every task run, but it's only actually used in the failure case. 
What do you think about adding a comment like 

`// Registry for actions that should be run if the task fails `

That would help future readers understand why it's unused on the success path.

Code after:
                }
                return;
            } else if (current == ExecutionState.CANCELING) {
                if (transitionState(ExecutionState.CANCELING, ExecutionState.CANCELED)) {
                    // we were immediately canceled. tell the TaskManager that we reached our final
                    // state
                    notifyFinalState();
                    if (metrics != null) {
                        metrics.close();
                    }
                    return;
                }
            } else {
                if (metrics != null) {
                    metrics.close();
                }
                throw new IllegalStateException(
                        "Invalid state for beginning of operation of task " + this + '.');
            }
        }

        // all resource acquisitions and registrations from here on
        // need to be undone in the end
        Map<String, Future<Path>> distributedCacheEntries = new HashMap<>();
        TaskInvokable invokable = null;
        // Registry that can be used to execute actions after the task has already failed. These
        // actions are fired in the registration order.
        AutoCloseableRegistry postFailureCleanUpRegistry = new AutoCloseableRegistry(false);

        try {
            // ----------------------------
            //  Task Bootstrap - We periodically
            //  check for canceling as a shortcut
            // ----------------------------

            // activate safety net for task thread
            LOG.debug("Creating FileSystem stream leak safety net for task {}", this);
            FileSystemSafetyNet.initializeSafetyNetForThread();

            // first of all, get a user-code classloader
            // this may involve downloading the job's JAR files and/or classes
            LOG.info("Loading JAR files for task {}.", this);

            userCodeClassLoader = createUserCodeClassloader();
            final ExecutionConfig executionConfig =
                    serializedExecutionConfig.deserializeValue(userCodeClassLoader.asClassLoader());
            Configuration executionConfigConfiguration = executionConfig.toConfiguration();

            // override task cancellation interval from Flink config if set in ExecutionConfig
            taskCancellationInterval =
                    executionConfigConfiguration