flink的checkpoint源码部分阅读一（Checkpoint接口）

大大源码 • 2023年4月26日 am12:01 • 未分类

对于checkpoint的各种理论看的有点多，但是感觉都不太深刻，所以就从码云上下载源码想多了解一下，可能有些理解不太准确，还望大佬们多多指正

前言

Checkpoint 使 Flink 的状态具有良好的容错性，通过 checkpoint 机制，Flink 可以对作业的状态和计算位置进行恢复。

了解Checkpoint

既然是阅读Checkpoint源码，那我们就得从Checkpoint这个接口（该接口位置在flink-runtime模块下，包名org.apache.flink.runtime.checkpoint）开始起步，如下：

/** A checkpoint, pending or completed.  检查点，待处理或已完成。 */
public interface Checkpoint {
    long getCheckpointID();

    void discard() throws Exception;
}

可以发现里面只有两个定义好的方法，一个是获得CheckpointID的函数，一个是discard函数。（由于关于这个接口的实现类的代码有点点多，所以我们接下来重点围绕这两个方法展开阅读）

该接口主要有两个实现类：
一个名叫CompletedCheckpoint类，一个名叫PendingCheckpoint类

CompletedCheckpoint

关于CompletedCheckpoint的源码注释是这样描述的：

CompletedCheckpoint 描述了在所有必需的任务都确认它（及其状态）并被认为成功之后的检查点。CompletedCheckpoint 类包含检查点的所有元数据，即检查点 ID、时间戳和作为检查点一部分的所有状态的句柄。

在大多数情况下，CompletedCheckpoint 对象非常小，因为检查点状态的句柄只是指针（例如文件路径）。但是，某些状态后端实现可能会选择将一些有效负载数据直接与元数据一起存储（例如以避免许多小文件）。如果这些阈值增加到较大的值，则 CompletedCheckpoint 对象的内存消耗可能会很大。

CompletedCheckpoint 的元数据也保存在外部存储系统中。检查点有一个指向元数据的外部指针。例如，在文件系统中存储检查点时，该指针是检查点文件夹或元数据文件的文件路径。对于将元数据存储在数据库表中的状态后端，指针可以是表名和行键。指针被编码为字符串。

关于那个getCheckpointID的实现，在这里是直接返回


    /** The ID (logical timestamp) of the checkpoint.
     * 检查点的 ID（逻辑时间戳）*/
    private final long checkpointID;

    @Override
    public long getCheckpointID() {
        return checkpointID;
    }

让我们再来看看discard函数的实现：

 @Override
    public void discard() throws Exception {
        LOG.trace("Executing discard procedure for {}.", this);

        try {
            // collect exceptions and continue cleanup 收集异常并继续清理
            Exception exception = null;

            // drop the metadata 删除元数据
            try {
                metadataHandle.discardState();
            } catch (Exception e) {
                exception = e;
            }

            // discard private state objects 丢弃私有状态对象
            try {
                StateUtil.bestEffortDiscardAllStateObjects(operatorStates.values());
            } catch (Exception e) {
                exception = ExceptionUtils.firstOrSuppressed(e, exception);
            }

            // discard location as a whole
            try {
                //配置存储位置
                storageLocation.disposeStorageLocation();
            } catch (Exception e) {
                exception = ExceptionUtils.firstOrSuppressed(e, exception);
            }

            if (exception != null) {
                throw exception;
            }
        } finally {
            //状态清除
            operatorStates.clear();

            // to be null-pointer safe, copy reference to stack 为了空指针安全，将引用复制到堆栈
            CompletedCheckpointStats.DiscardCallback discardCallback = this.discardCallback;
            if (discardCallback != null) {
                discardCallback.notifyDiscardedCheckpoint();
            }
        }
    }

在什么情况下会触发这个discard函数呢？
源码如下：
在作业结束，取消，失败，暂停时触发discard函数
关于作业的调度请看–>flink-作业调度

    public boolean shouldBeDiscardedOnShutdown(JobStatus jobStatus) {
        //在作业 结束，取消，失败，暂停时触发丢弃函数
        return jobStatus == JobStatus.FINISHED && props.discardOnJobFinished()
                || jobStatus == JobStatus.CANCELED && props.discardOnJobCancelled()
                || jobStatus == JobStatus.FAILED && props.discardOnJobFailed()
                || jobStatus == JobStatus.SUSPENDED && props.discardOnJobSuspended();
    }

PendingCheckpoint

关于CompletedCheckpoint的源码注释是这样描述的：

挂起的检查点是已启动但尚未被所有需要确认它的任务确认的检查点。一旦所有任务都确认了它，它就会变成一个 {@link CompletedCheckpoint}。

请注意，挂起的检查点以及成功的检查点将状态句柄始终作为序列化值处理，而不是作为实际值。

关于那个getCheckpointID的实现，在这里也是直接返回

    /** @deprecated use {@link #getCheckpointID()} */
    @Deprecated
    public long getCheckpointId() {
        return getCheckpointID();
    }

    @Override
    public long getCheckpointID() {
        return checkpointId;
    }

让我们再来看看discard函数的实现：
discard 状态。必须在 {@link #dispose(boolean, CheckpointsCleaner, Runnable, Executor) dispose} 之后调用。

   /**
     * Discard state. Must be called after {@link #dispose(boolean, CheckpointsCleaner, Runnable,
     * Executor) dispose}.
     * 丢弃状态。必须在 {@link #dispose(boolean, CheckpointsCleaner, Runnable, Executor) dispose} 之后调用。
     */
    @Override
    public void discard() {
        synchronized (lock) {
            if (discarded) {
                Preconditions.checkState(
                        disposed, "Checkpoint should be disposed before being discarded");
                return;
            } else {
                discarded = true;
            }
        }
        // discard the private states. 丢弃私有状态
        // unregistered shared states are still considered private at this point. 未注册的共享状态此时仍被认为是私有的。
        try {
            //尽最大努力丢弃所有状态对象
            StateUtil.bestEffortDiscardAllStateObjects(operatorStates.values());
            //失败时处理
            targetLocation.disposeOnFailure();
        } catch (Throwable t) {
            LOG.warn(
                    "Could not properly dispose the private states in the pending checkpoint {} of job {}.",
                    checkpointId,
                    jobId,
                    t);
        } finally {
        //清除算子状态
            operatorStates.clear();
        }
    }