Flink 1.5版本的“Task本地恢复”实现摘要

Posted 2022-12-06 vinoYang

overview

解读Flink 1.5版本的“task本地恢复”feature的实现。

随着时间的推移，Flink有些模块的代码复杂度已经很高了（比较典型的就是状态&容错这块）。因为考虑到兼容历史实现的问题，类在快速增长。

实现

关于本地恢复的配置，封装在名为LocalRecoveryConfig的类中，该类可提供本地恢复模式的查询（也即LocalRecoveryMode枚举）以及本地恢复目录提供者（LocalRecoveryDirectoryProvider）的获取。

LocalRecoveryMode 的枚举值有两个：DISABLED 以及 ENABLE_FILE_BASED。可以从检查点的特定的字符串配置中进行转换，相关配置项如下：

    /**
     * This option configures local recovery for this state backend.
     */
    public static final ConfigOption<String> LOCAL_RECOVERY = ConfigOptions
        .key("state.backend.local-recovery")
        .defaultValue("DISABLED");

除此之外，还有一个跟用户相关的配置项，用于存放本地状态的根目录：

    /**
     * The config parameter defining the root directories for storing file-based state for local recovery.
     */
    public static final ConfigOption<String> LOCAL_RECOVERY_TASK_MANAGER_STATE_ROOT_DIRS = ConfigOptions
        .key("taskmanager.state.local.root-dirs")
        .noDefaultValue();

TaskLocalStateStore是任务本地状态的核心接口，TaskLocalStateStoreImpl是其唯一的实现，我们通过其构造方法来看一下，需要哪些信息才能确认或构建出一个TaskLocalStateStore:

    public TaskLocalStateStoreImpl(
        @Nonnull JobID jobID,
        @Nonnull AllocationID allocationID,
        @Nonnull JobVertexID jobVertexID,
        @Nonnegative int subtaskIndex,
        @Nonnull LocalRecoveryConfig localRecoveryConfig,
        @Nonnull Executor discardExecutor) 

在TaskManager或TaskExecutor中submitTask时需要查找当前subTask的TaskLocalStateStore。该查找由TaskExecutorLocalStateStoresManager的如下方法提供：

public TaskLocalStateStore localStateStoreForSubtask(
        @Nonnull JobID jobId,
        @Nonnull AllocationID allocationID,
        @Nonnull JobVertexID jobVertexID,
        @Nonnegative int subtaskIndex)

TaskExecutorLocalStateStoresManager用于管理某task executor中所有的TaskLocalStateStoreImpl（通过一个<JobVertexSubtaskKey, TaskLocalStateStoreImpl>的Map来存储其对应关系）。

在localStateStoreForSubtask方法中，如果没有查找到对应的TaskLocalStateStore会为其初始化，该初始化的过程包括了创建本地恢复目录提供者LocalRecoveryDirectoryProviderImpl（主要用于为任务实例分配、选择目录），上面的根目录配置就是为它所用。

TaskStateManager以及TaskStateManagerImpl充当了检查点的reporter（除此之外，还包括了代理LocalRecoveryConfig的创建），它用于向外部report检查点的相关信息，其跟具体的sub-task以及TaskLocalStore是一对一的关系，并且会随着Environment 穿透到上层StreamTask中去。

snapshot

在特定的KeyedStateBackend中，会根据是否启用了本地恢复来创建不同的Checkpoint输出流Provider。具体而言，如果启用了本地恢复，那么将会创建一个兼具“主要CP”以及“次级CP”的流，名为PrimaryAndSecondaryStream（具体的次级CP输出流的实现为基于文件的FileBasedStateOutputStream）。而如果没有启用本地恢复，则只会创建一个PrimaryStreamOnly（它即为Flink之前版本中的CP实现）。

recovery

我们都知道在submitTask调用中会构建TaskLocalStateStore的实例，而其retrieveLocalState主要用于在恢复时，做状态重置时获取状态。我们来看一下TaskLocalStateStoreImpl.retrieveLocalState的调用栈，如下图：

这里引入了两个类：PrioritizedOperatorSubtaskState 以及 BackendRestorerProcedure.

PrioritizedOperatorSubtaskState封装了某个subTask的多个快照，从类名可以看出来它引入了优先级机制，如果某个subTask既存在从JM端获取的快照又存在本地文件系统的快照，本地文件系统的快照将作为候选。

具体确定谁优先级最高，可参见该方法的实现：PrioritizedOperatorSubtaskState.Builder.resolvePrioritizedAlternatives。基本上如果本地有快照，那么采用本地优先的策略。

针对 ” 优先级，多候选 ” 的恢复方式，提供了BackendRestorerProcedure来实现恢复，它会按照优先级顺序，进行恢复，直到成功。具体逻辑见BackendRestorerProcedure.createAndRestore以及BackendRestorerProcedure.attemptCreateAndRestore。