Checkpoint的触发方式有两种

• 一种是数据源节点中的Checkpoint操作触发，通过CheckpointCoordinator组件进行协调和控制。 CheckpointCoordinator通过注册定时器的方式按照配置的时间间隔触发数据源节点的Checkpoint操作。数据源节点会向下游算子发出Checkpoint Barrier事件，供下游节点使用。
• 另一种是下游算子节点根据上游发送的Checkpoint Barrier事件控制算子中Checkpoint操作的触发时机，即只有接收到所有上游Barrier事件后，才会触发本节点的Checkpoint操作。

本文先介绍通过CheckpointCoordinator触发算子的Checkpoint操作

CheckpointCoordinator在整个作业中扮演了Checkpoint协调者的角色，负责在数据源节点触发Checkpoint以及整个作业的Checkpoint管理，并且CheckpointCoordinator组件会接收TaskMananger在Checkpoint执行完成后返回的Ack消息。

 

一. 启动CheckpointCoordinator

当作业的JobStatus转换为Running时，通知CheckpointCoordinatorDeActivator监听器启动CheckpointCoordinator服务。

如代码CheckpointCoordinatorDeActivator.jobStatusChanges()方法主要包含如下逻辑。

> 1. 当`newJobStatus == JobStatus.RUNNING`时，立即调用
> coordinator.startCheckpointScheduler()方法启动整个Job的调度器
> CheckpointCoordinator，此时Checkpoint的触发依靠CheckpointCoordinator进行协调。
> 
> 2. 当JobStatus为其他类型状态时，调用coordinator.stopCheckpointScheduler()方法，
> 停止当前Job中的Checkpoint操作。public class CheckpointCoordinatorDeActivator implements JobStatusListener {private final CheckpointCoordinator coordinator;public CheckpointCoordinatorDeActivator(CheckpointCoordinator coordinator) {this.coordinator = checkNotNull(coordinator);}@Overridepublic void jobStatusChanges(JobID jobId,JobStatus newJobStatus, long timestamp,Throwable error) {if (newJobStatus == JobStatus.RUNNING) {// 启动Checkpoint调度程序coordinator.startCheckpointScheduler();} else {// 直接停止CheckpointSchedulercoordinator.stopCheckpointScheduler();}}
}

 

二. 开启CheckpointScheduler线程

接下来在CheckpointCoordinator.startCheckpointScheduler()方法中调用scheduleTriggerWithDelay()方法进行后续操作，向创建好的checkpointCoordinatorTimer线程池添加定时调度执行的Runnable线程。

如代码所示：

在CheckpointCoordinator.scheduleTriggerWithDelay()方法中指定baseInterval参数，设定执行Checkpoint操作的时间间隔，通过定时器周期性地触发ScheduledTrigger线程，Checkpoint的具体操作在ScheduledTrigger线程中实现。

private ScheduledFuture<?> scheduleTriggerWithDelay(long initDelay) {return timer.scheduleAtFixedRate(new ScheduledTrigger(),initDelay, baseInterval, TimeUnit.MILLISECONDS);
}

 

三. 触发Checkpoint

如代码，ScheduledTrigger也是CheckpointCoordinator的内部类，实现了Runnable接口。在ScheduledTrigger.run()方法中调用了CheckpointCoordinator.triggerCheckpoint()方法触发和执行Checkpoint操作。

private final class ScheduledTrigger implements Runnable {@Overridepublic void run() {try {// 调用triggerCheckpoint()方法触发Checkpoint操作triggerCheckpoint(System.currentTimeMillis(), true);}catch (Exception e) {LOG.error("Exception while triggering checkpoint for job {}.", job, e);}}
}

CheckpointCoordinator.triggerCheckpoint()方法包含的执行逻辑非常多，这里重点介绍其中的主要逻辑。根据CheckpointCoordinator触发Checkpoint操作的过程分为以下几个部分。

1. Checkpoint执行前的工作

1. 首先检查Checkpoint的执行环境和参数，满足条件后触发执行Checkpoint操作。Checkpoint执行过程分为异步和同步两种：

调用preCheckBeforeTriggeringCheckpoint()方法进行一些前置检查，主要包括检查CheckpointCoordinator当前的状态是否为shutdown、Checkpoint尝试次数是否超过配置的最大值。

2. 构建执行和触发Checkpoint操作对应的Task节点实例的Execution集合，其中tasksToTrigger数组中存储了触发Checkpoint操作的ExecutionVertex元素，实际上就是所有的数据源节点。

CheckpointCoordinator仅会触发数据源节点的Checkpoint操作，其他节点则是通过Barrier对齐的方式触发的。

3. 构建需要发送Ack消息的ExecutionVertex集合，主要是从tasksToWaitFor集合中转换而来。

tasksToWaitFor中存储了ExecutonGraph中所有的ExecutionVertex，也就是说每个ExecutionVertex节点对应的Task实例都需要向CheckpointCoordinator中汇报Ack消息。

// 主要做前置检查synchronized (lock) {preCheckBeforeTriggeringCheckpoint(isPeriodic, props.forceCheckpoint());}// 创建需要执行的Task对应的Execution集合Execution[] executions = new Execution[tasksToTrigger.length];// 遍历tasksToTrigger集合，构建Execution集合for (int i = 0; i < tasksToTrigger.length; i++) {//获取Task对应的Execution集合Execution ee = tasksToTrigger[i].getCurrentExecutionAttempt();if (ee == null) {// 如果Task对应的Execution集合为空，代表Task没有被执行，则抛出异常LOG.info("Checkpoint triggering task {} of job {} is not being executed at the moment. Aborting checkpoint.", tasksToTrigger[i].getTaskNameWithSubtaskIndex(), job);throw new CheckpointException(CheckpointFailureReason.NOT_ALL_REQUIRED_TASKS_RUNNING);} else if (ee.getState() == ExecutionState.RUNNING) {// 如果ExecutionState为RUNNING，则添加到executions集合中executions[i] = ee;} else {// 如果其他ExecutionState不为RUNNING，则抛出异常LOG.info("Checkpoint triggering task {} of job {} is not in state {} but {} instead. Aborting checkpoint.",tasksToTrigger[i].getTaskNameWithSubtaskIndex(),job,ExecutionState.RUNNING,ee.getState());throw new CheckpointException(CheckpointFailureReason.NOT_ALL_REQUIRED_TASKS_RUNNING);}}// 组装用于需要发送Ack消息的Task集合Map<ExecutionAttemptID, ExecutionVertex> ackTasks = new HashMap<>(tasksToWaitFor.length);for (ExecutionVertex ev : tasksToWaitFor) {Execution ee = ev.getCurrentExecutionAttempt();if (ee != null) {ackTasks.put(ee.getAttemptId(), ev);} else {LOG.info("Checkpoint acknowledging task {} of job {} is not being executed at the moment. Aborting checkpoint.", ev.getTaskNameWithSubtaskIndex(), job);throw new CheckpointException(CheckpointFailureReason.NOT_ALL_REQUIRED_TASKS_RUNNING);}
}

 

2. 创建PendingCheckpoint

在执行Checkpoint操作之前，需要构建PendingCheckpoint对象，从字面意思上讲就是挂起Checkpoint操作。

从开始执行Checkpoint操作直到Task实例返回Ack确认成功消息，Checkpoint会一直处于Pending状态，确保Checkpoint能被成功执行。

如代码逻辑：

1. Checkpoint有唯一的checkpointID标记，根据高可用模式选择不同的计数器。

如果基于ZooKeeper实现了高可用集群，会调用ZooKeeperCheckpointIDCounter实现checkpointID计数；如果是非高可用集群，则会通过StandaloneCheckpointIDCounter完成checkpointID计数。

2. 创建checkpointStorageLocation，用于定义Checkpoint过程中状态快照数据存放的位置。

checkpointStorageLocation通过checkpointStorage创建和初始化，不同的checkpointStorage实现创建的checkpointStorageLocation会有所不同。

3. 创建PendingCheckpoint对象。

包括checkpointID、ackTasks以及checkpointStorageLocation等参数信息。将创建好的PendingCheckpoint存储在pendingCheckpoints集合中，并异步执行PendingCheckpoint操作。

final CheckpointStorageLocation checkpointStorageLocation;
final long checkpointID;
try {//通过checkpointIdCounter获取checkpointIDcheckpointID = checkpointIdCounter.getAndIncrement();// 获取checkpointStorageLocationcheckpointStorageLocation = props.isSavepoint() ?checkpointStorage.initializeLocationForSavepoint(checkpointID, externalSavepointLocation) :checkpointStorage.initializeLocationForCheckpoint(checkpointID);
}
// 省略部分代码
// 创建PendingCheckpoint对象
final PendingCheckpoint checkpoint = new PendingCheckpoint(job,checkpointID,timestamp,ackTasks,masterHooks.keySet(),props,checkpointStorageLocation,executor);

 

3. Checkpoint的触发与执行

在CheckpointCoordinator.triggerCheckpoint()方法中，会在synchronized(lock)模块内定义和执行Checkpoint操作的具体逻辑，主要包含如下步骤。

1. 获取coordinator对象锁，对TriggeringCheckpoint对象进行预检查，主要包括检查CheckpointCoordinator状态和PendingCheckpoint尝试次数等。

2. 将PendingCheckpoint存储在pendingCheckpoints键值对中，使用定时器创建cancellerHandle对象，cancellerHandle用于清理过期的Checkpoint操作。

通过checkpoint.setCancellerHandle()方法设置Checkpoint的CancellerHandle，设置成功则返回True，如果失败则返回false，说明当前Checkpoint已经被释放。

3. 调用并执行MasterHook。可以通过实现MasterHook函数，准备外部系统环境或触发相应的系统操作。

4. 遍历执行executions集合中的Execution节点，判断props.isSynchronous()方法是否为True，如果为True则调用triggerSynchronousSavepoint()方法同步执行Checkpoint操作。
   其他情况则调用triggerCheckpoint()方法异步执行Checkpoint操作。

// 获取coordinator-wide lock
synchronized (lock) {// TriggeringCheckpoint检查preCheckBeforeTriggeringCheckpoint(isPeriodic, props.forceCheckpoint());LOG.info("Triggering checkpoint {} @ {} for job {}.", checkpointID, timestamp, job);// 将checkpoint存储在pendingCheckpoints KV集合中pendingCheckpoints.put(checkpointID, checkpoint);// 调度canceller线程，清理过期的Checkpoint对象ScheduledFuture<?> cancellerHandle = timer.schedule(canceller,checkpointTimeout, TimeUnit.MILLISECONDS);// 确定Checkpoint是否已经被释放if (!checkpoint.setCancellerHandle(cancellerHandle)) {cancellerHandle.cancel(false);}// 调用MasterHook方法for (MasterTriggerRestoreHook<?> masterHook : masterHooks.values()) {final MasterState masterState =MasterHooks.triggerHook(masterHook, checkpointID, timestamp, executor).get(checkpointTimeout, TimeUnit.MILLISECONDS);checkpoint.acknowledgeMasterState(masterHook.getIdentifier(), masterState);}Preconditions.checkState(checkpoint.areMasterStatesFullyAcknowledged());
}
// 创建CheckpointOptions
final CheckpointOptions checkpointOptions = new CheckpointOptions(props.getCheckpointType(),checkpointStorageLocation.getLocationReference());
// 分别执行executions中的Execution节点
for (Execution execution: executions) {if (props.isSynchronous()) {// 如果是同步执行，则调用triggerSynchronousSavepoint()方法execution.triggerSynchronousSavepoint(checkpointID, timestamp, checkpointOptions,advanceToEndOfTime);} else {// 其他情况则调用triggerCheckpoint()异步方法执行execution.triggerCheckpoint(checkpointID, timestamp, checkpointOptions);}
}
// 返回Checkpoint中的CompletionFuture对象
numUnsuccessfulCheckpointsTriggers.set(0);
return checkpoint.getCompletionFuture();

以上就完成了在CheckpointCoordinator中触发Checkpoint的全部操作，具体的执行过程调用Execution完成。

 

四. Task节点的Checkpoint操作

在Execution.triggerCheckpoint()方法中实际上调用triggerCheckpointHelper()方法完成Execution对应的Task节点的Checkpoint操作，并通过Task实例触发数据源节点的Checkpoint操作，如代码所示。

1. 触发准备

1. 获取当前Execution分配的LogicalSlot，如果LogicalSlot不为空，说明Execution成功分配到Slot计算资源，否则说明Execution中没有资源，Execution对应的Task实例不会被执行和启动。

2. 调用TaskManagerGateway.triggerCheckpoint()的RPC方法，触发和执行指定Task的Checkpoint操作。

3. TaskExecutor收到来自CheckpointCoordinator的Checkpoint触发请求后，会在TaskExecutor实例中完成对应Task实例的Checkpoint操作。

private void triggerCheckpointHelper(long checkpointId, long timestamp, CheckpointOptions checkpointOptions, boolean advanceToEndOfEventTime) {final CheckpointType checkpointType = checkpointOptions.getCheckpointType();if (advanceToEndOfEventTime && !(checkpointType.isSynchronous() && checkpointType.isSavepoint())) {throw new IllegalArgumentException("Only synchronous savepoints are allowed to advance the watermark to MAX.");}// 获取当前Execution分配的LogicalSlot资源final LogicalSlot slot = assignedResource;// 如果LogicalSlot不为空，说明Execution运行正常if (slot != null) {// 通过slot获取TaskManagerGateway对象final TaskManagerGateway taskManagerGateway = slot.getTaskManagerGateway();// 调用triggerCheckpoint()方法taskManagerGateway.triggerCheckpoint(attemptId, getVertex().getJobId(), checkpointId, timestamp, checkpointOptions,advanceToEndOfEventTime);} else {// 否则说明Execution中没有资源，不再执行Execution对应的Task实例LOG.debug("The execution has no slot assigned. This indicates that the execution is no longer running.");}
}

 

2. 调用TaskExecutor执行Checkpoint操作

TaskExecutor接收到来自CheckpointCoordinator的Checkpoint触发请求后，立即根据Execution信息确认Task实例线程，并且调用Task实例触发和执行数据源节点的Checkpoint操作。如代码，TaskExecutor.triggerCheckpoint()方法逻辑如下。

1. 检查CheckpointType的类型，CheckpointType共有三种类型，分别为CHECKPOINT、SAVEPOINT和SYNC_SAVEPOINT，且只有在同步Savepoints操作时才能调整Watermark为MAX。

2. 从taskSlotTable中获取Execution对应的Task实例，如果Task实例不为空，则调用task.triggerCheckpointBarrier()方法执行Task实例中的Checkpoint操作。

3. 如果Task实例为空，说明Task目前处于异常，无法执行Checkpoint操作。此时调用FutureUtils.completedExceptionally()方法，并封装CheckpointException异常信息，返回给管理节点的CheckpointCoordinator进行处理。

public CompletableFuture<Acknowledge> triggerCheckpoint(ExecutionAttemptID executionAttemptID,long checkpointId,long checkpointTimestamp,CheckpointOptions checkpointOptions,boolean advanceToEndOfEventTime) {log.debug("Trigger checkpoint {}@{} for {}.", checkpointId, checkpointTimestamp, executionAttemptID);//检查CheckpointType，确保只有同步的savepoint操作才能将Watermark调整为MAXfinal CheckpointType checkpointType = checkpointOptions.getCheckpointType();if (advanceToEndOfEventTime && !(checkpointType.isSynchronous() && checkpointType.isSavepoint())) {throw new IllegalArgumentException("Only synchronous savepoints are allowed to advance the watermark to MAX.");}// 从taskSlotTable中获取当前Execution对应的Taskfinal Task task = taskSlotTable.getTask(executionAttemptID);// 如果task不为空，则调用triggerCheckpointBarrier()方法if (task != null) {task.triggerCheckpointBarrier(checkpointId, checkpointTimestamp, checkpointOptions, advanceToEndOfEventTime);// 返回CompletableFuture对象return CompletableFuture.completedFuture(Acknowledge.get());} else {final String message = "TaskManager received a checkpoint request for unknown task " + executionAttemptID + '.';// 如果task为空，则返回CheckpointException异常log.debug(message);return FutureUtils.completedExceptionally(new CheckpointException(message,
CheckpointFailureReason.TASK_CHECKPOINT_FAILURE));}
}

 

五. 在StreamTask中执行Checkpoint操作

在执行Task.triggerCheckpointBarrier()方法时，会借助AbstractInvokable中提供的triggerCheckpointAsync()方法触发并执行StreamTask中的Checkpoint操作。

public Future<Boolean> triggerCheckpointAsync(CheckpointMetaData checkpointMetaData,CheckpointOptions checkpointOptions,boolean advanceToEndOfEventTime) {// 异步提交Checkpoint操作return mailboxProcessor.getMainMailboxExecutor().submit(    () -> triggerCheckpoint(checkpointMetaData, checkpointOptions, advanceToEndOfEventTime),"checkpoint %s with %s",checkpointMetaData,checkpointOptions);
}

StreamTask.triggerCheckpoint()方法主要逻辑如下。

1. 调用StreamTask.performCheckpoint()方法执行Checkpoint并返回success信息，用于判断Checkpoint操作是否成功执行。
2. 如果success信息为False，表明Checkpoint操作没有成功执行，此时调用declineCheckpoint()方法回退。

boolean success = performCheckpoint(checkpointMetaData, checkpointOptions, checkpointMetrics, advanceToEndOfEventTime);
if (!success) {declineCheckpoint(checkpointMetaData.getCheckpointId());
}
return success;

在StreamTask.performCheckpoint()方法中，主要执行了Task实例的Checkpoint操作，该方法除了会通过CheckpointCoordinator触发之外，在下游算子通过CheckpointBarrier对齐触发Checkpoint操作时，也会调用该方法执行具体Task的Checkpoint操作。

 
下篇我们继续看CheckpointBarrier对齐触发Checkpoint的流程，了解StreamTask中performCheckpoint()方法如何执行Checkpoint操作，实现状态数据快照与持久化操作。

 

参考：《Flink设计与实现：核心原理与源码解析》–张利兵