Flink Checkpoint 源码（二）：Barrier 对齐与状态快照

前些天发现了一个巨牛的人工智能学习网站，通俗易懂，风趣幽默，忍不住给大家分享一下。点击跳转到网站：https://www.captainai.net/dongkelun

前言

上一篇分析了 JobManager 端 CheckpointCoordinator 的触发链路：从定时触发、请求决策，到发送触发 RPC、收齐确认、完成 Checkpoint。这一篇转到 TaskManager 端，回答上一篇留下的问题：Task 收到触发请求或 Barrier 之后，如何对齐 Barrier、如何执行状态快照、如何把确认发回 JobManager。

Task 端一次 Checkpoint 的完整路径：

1
Source 注入 Barrier → Barrier 随数据流传播 → 中间 Task 对齐 → 状态快照（同步 + 异步）→ 向 JobManager 发 ack

前置知识：

版本

Flink 1.15.3

一、Task 端概览

Task 端的核心组件：

1
2
3
4
5
6
7
8
9
10
StreamTask（mailbox 线程）
  ├── SubtaskCheckpointCoordinator    快照协调器
  │     └── AsyncCheckpointRunnable   异步阶段（asyncOperationsThreadPool）
  ├── CheckpointBarrierHandler        Barrier 状态机
  │     ├── SingleCheckpointBarrierHandler   aligned / alternating / unaligned
  │     └── CheckpointBarrierTracker         AT_LEAST_ONCE（只计数不阻塞）
  ├── CheckpointedInputGate           网络层 Barrier 事件分发
  ├── OperatorChain                   算子链
  │     └── StreamOperatorStateHandler  单算子快照
  └── StateBackend                    keyed 状态 / operator 状态

分工：

  • CheckpointBarrierHandler 负责”何时快照”——接收各输入通道的 Barrier，执行对齐，到齐后通知 Task 快照
  • SubtaskCheckpointCoordinator 负责”怎么快照”——执行快照的同步阶段和异步阶段，把结果上报 JobManager
  • StreamOperatorStateHandler 负责”单个算子怎么快照”——调用用户算子的 snapshotState() 钩子和 StateBackend 的快照

mailbox 线程是 StreamTask 处理数据的唯一线程：数据记录、watermark、控制事件都提交到 mailbox 队列，在这个线程里串行执行。Checkpoint 同步阶段也是提交到 mailbox 执行的，因此天然和数据处理串行。

线程模型是读这篇的主线：

阶段 线程 内容
Barrier 处理（对齐） mailbox 从网络读 Barrier、阻塞/放行通道
快照同步阶段 mailbox prepareSnapshotPreBarrier、转发 Barrier、各算子快照（拿到 future）
快照异步阶段 asyncOperationsThreadPool 等待 future 完成、状态落盘、发 ack

同步阶段为什么必须在 mailbox 线程？Barrier 是数据流上的切割点，快照必须相对数据流”原子”——发生在两条记录之间，不能和数据处理交错。mailbox 线程是处理数据的唯一线程，同步阶段放在这里才能保证原子性。异步阶段是磁盘 IO，耗时不可控，放到线程池避免阻塞数据流。

二、Barrier 的注入与到达

2.1 哪些 Task 被 RPC 触发

先澄清一个容易误解的点：触发 RPC 不是发给所有 Task 的。

CheckpointCoordinator 计算 CheckpointPlan 时，calculateWithAllTasksRunning() 里 tasksToTrigger = sourceTasks——只有 Source Task 收到触发 RPC。中间 Task 不靠 RPC 触发，靠 Barrier 本身经数据流到达。

这是 Chandy-Lamport 算法的起点：协调者只需要告诉 Source”请注入 Barrier”，剩下的由 Barrier 的传播完成。

2.2 Source Task：Barrier 如何进入数据流

Source Task 收到 RPC 后的路径：

1
2
3
4
5
6
7
TaskExecutor.triggerCheckpoint()          ← RPC 入口
  └─ Task.triggerCheckpointBarrier()
       └─ StreamTask.triggerCheckpointAsync()
            └─ [mailbox] triggerCheckpointAsyncInMailbox()
                 └─ performCheckpoint()
                      └─ SubtaskCheckpointCoordinatorImpl.checkpointState()
                           └─ operatorChain.broadcastEvent(CheckpointBarrier)   ← Barrier 注入

StreamTask.triggerCheckpointAsync() 区分两种情况：

1
2
3
4
5
6
7
8
9
10
11
12
// StreamTask.java
boolean noUnfinishedInputGates =
        Arrays.stream(getEnvironment().getAllInputGates())
                .allMatch(InputGate::isFinished);

if (noUnfinishedInputGates) {
    // 无网络输入（Source）→ 直接执行快照
    result.complete(triggerCheckpointAsyncInMailbox(checkpointMetaData, checkpointOptions));
} else {
    // 还有未结束的输入通道 → 对这些通道手动调 processBarrier
    result.complete(triggerUnfinishedChannelsCheckpoint(checkpointMetaData, checkpointOptions));
}

InputGate 是网络层接收上游数据的组件：上游每个 subtask 的输出分区对应一个 InputChannel，多个 channel 汇聚到一个 InputGate，Task 从这里读上游发来的数据。Source Task 没有上游，数据来自外部系统（Kafka、文件等），不走网络，所以没有 InputGate。

纯 Source 没有输入 gate，getAllInputGates() 返回空、allMatch 对空集合返回 true，走第一个分支直接执行快照。Barrier 在 checkpointState() 的第 (2) 步注入数据流：

1
2
3
4
5
// SubtaskCheckpointCoordinatorImpl.java
// Step (2): Send the checkpoint barrier downstream
operatorChain.broadcastEvent(
        new CheckpointBarrier(metadata.getCheckpointId(), metadata.getTimestamp(), options),
        options.isUnalignedCheckpoint());

Barrier 是随数据流传输的特殊事件（RuntimeEvent），和数据严格保序。这是 Chandy-Lamport 算法的基础——Barrier 把数据流切成两段：Barrier 之前的数据属于本次 Checkpoint 要快照的状态，之后的属于下一个 epoch。

第二个参数 options.isUnalignedCheckpoint() 在非对齐模式下为 true：Barrier 作为优先事件写入输出，越过在途数据（见第四节）。

第二个分支（triggerUnfinishedChannelsCheckpoint）是边界场景：上游部分结束、Task 收不齐网络 Barrier 时，对未结束的通道手动调用 processBarrier(barrier, channelInfo, true)，isRpcTriggered=true 表示该 Barrier 由 RPC 触发，不阻塞通道。

2.3 中间 Task：Barrier 如何到达

Barrier 经网络层到达中间 Task，CheckpointedInputGate 是事件入口：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
// CheckpointedInputGate.java
private Optional<BufferOrEvent> handleEvent(BufferOrEvent bufferOrEvent) throws IOException {
    Class<? extends AbstractEvent> eventClass = bufferOrEvent.getEvent().getClass();
    if (eventClass == CheckpointBarrier.class) {
        // 对齐/非对齐处理入口，isRpcTriggered=false
        barrierHandler.processBarrier(checkpointBarrier, bufferOrEvent.getChannelInfo(), false);
    } else if (eventClass == CancelCheckpointMarker.class) {
        // 上游 abort 后广播的取消标记，通知下游别等
        barrierHandler.processCancellationBarrier(...);
    } else if (eventClass == EndOfPartitionEvent.class) {
        // 上游分区结束
        barrierHandler.processEndOfPartition(...);
    } else if (eventClass == EventAnnouncement.class) {
        // Barrier 即将到达的预告（非对齐切换时用于定位 Barrier）
        barrierHandler.processBarrierAnnouncement(...);
    }
    ...
}

和 Checkpoint 相关的事件有四种：

  • CheckpointBarrier：Barrier 本身，进入对齐状态机
  • CancelCheckpointMarker：取消标记，上游 abort 后广播，让下游别等这个 Barrier
  • EndOfPartitionEvent：上游分区结束（”任务结束后仍可 Checkpoint”场景使用）
  • EventAnnouncement：Barrier 到达预告，网络层提前通知，alternating 模式超时切换非对齐时靠它定位 Barrier 的位置

2.4 选择哪个 Handler

Barrier 处理器在 Task 的 init 阶段创建，InputProcessorUtil.createCheckpointBarrierHandler() 按 Checkpoint 模式选择：

Checkpoint 模式 非对齐开关 Handler 初始状态
EXACTLY_ONCE 关 SingleCheckpointBarrierHandler.aligned WaitingForFirstBarrier
EXACTLY_ONCE 开 SingleCheckpointBarrierHandler.alternating AlternatingWaitingForFirstBarrier
AT_LEAST_ONCE — CheckpointBarrierTracker 无状态机，只计数

具体对齐模式由 CheckpointOptions.forConfig() 决定（JobManager 端 triggerTasks 构造，随 Barrier 下发）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
// CheckpointOptions.java
public static CheckpointOptions forConfig(...) {
    if (!isExactlyOnceMode) {
        return notExactlyOnce(...);              // AT_LEAST_ONCE：不对齐
    } else if (checkpointType.isSavepoint()) {
        return alignedNoTimeout(...);            // Savepoint：对齐，无超时
    } else if (!isUnalignedEnabled) {
        return alignedNoTimeout(...);            // 未开非对齐：纯对齐
    } else if (alignedCheckpointTimeout == 0) {
        return unaligned(...);                   // 超时为 0：直接非对齐
    } else {
        return alignedWithTimeout(...);          // 超时 > 0：先对齐，超时切非对齐
    }
}

四种对齐模式：

  • ALIGNED：纯对齐，等所有通道 Barrier 到齐，无超时
  • ALIGNED + 超时（alternating）：先对齐，超时未到齐切非对齐
  • UNALIGNED：直接非对齐，第一个 Barrier 即触发快照
  • AT_LEAST_ONCE：不对齐，只计数（CheckpointBarrierTracker 不阻塞通道，无 exactly-once 保证）

Savepoint 一律对齐模式，不支持非对齐。

三、对齐 Checkpoint：状态机

3.1 状态机结构

对齐 Checkpoint 用状态机实现。SingleCheckpointBarrierHandler 持有当前状态 currentState，每个事件驱动一次状态转换：

1
2
3
4
5
6
7
8
9
10
WaitingForFirstBarrier ──第一个 Barrier──► CollectingBarriers
（等待第一个 Barrier）                      （收集 Barrier，已阻塞部分通道）
                                                  │
                                                  │ 所有通道到齐
                                                  ▼
                                        triggerGlobalCheckpoint
                                        （触发快照 + 放行所有通道）
                                                  │
                                                  ▼
                                        WaitingForFirstBarrier（回到初始）

状态机接口 BarrierHandlerState 有五个事件方法，每个都返回下一个状态：

1
2
3
4
5
6
7
8
// BarrierHandlerState.java
interface BarrierHandlerState {
    BarrierHandlerState barrierReceived(...);        // 收到 Barrier
    BarrierHandlerState announcementReceived(...);   // 收到 Barrier 预告
    BarrierHandlerState alignmentTimeout(...);       // 对齐超时
    BarrierHandlerState abort(long cancelledId);     // 中止
    BarrierHandlerState endOfPartitionReceived(...); // 上游分区结束
}

状态机通过 Controller 接口与 Handler 交互（触发快照、判断是否到齐、取 pending Barrier 等）。通道的阻塞/放行状态记录在 ChannelState：

  • blockedChannels：已处理 Barrier 且被阻塞的通道
  • sequenceNumberInAnnouncedChannels：已收到预告但 Barrier 还没到的通道

3.2 processBarrier：过滤与新 Checkpoint

processBarrier() 是 Barrier 处理的统一入口：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
// SingleCheckpointBarrierHandler.java
public void processBarrier(CheckpointBarrier barrier, InputChannelInfo channelInfo, boolean isRpcTriggered)
        throws IOException {
    long barrierId = barrier.getId();

    // 旧 Barrier（id 更小，或同 id 但 Checkpoint 已完成）→ 忽略
    if (currentCheckpointId > barrierId
            || (currentCheckpointId == barrierId && !isCheckpointPending())) {
        if (!barrier.getCheckpointOptions().isUnalignedCheckpoint()) {
            inputs[channelInfo.getGateIdx()].resumeConsumption(channelInfo);
        }
        return;
    }

    checkNewCheckpoint(barrier);
    checkState(currentCheckpointId == barrierId);

    markCheckpointAlignedAndTransformState(
            channelInfo, barrier,
            state -> state.barrierReceived(context, channelInfo, barrier, !isRpcTriggered));
}

做三件事：

  1. 过滤旧 Barrier。Barrier id 单调递增，id 更小的属于旧 Checkpoint（可能已完成或已 abort），直接忽略
  2. checkNewCheckpoint()：如果是新 Checkpoint 的第一个 Barrier，且旧 Checkpoint 还没完成，先 abort 旧的（subsume，见 3.5）
  3. markCheckpointAlignedAndTransformState()：记录已对齐通道、对齐计时、驱动状态机转换

markCheckpointAlignedAndTransformState() 负责对齐计时：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
// SingleCheckpointBarrierHandler.java
alignedChannels.add(alignedChannel);
if (alignedChannels.size() == 1) {
    // 第一个 Barrier 到达 → 开始对齐计时
    markAlignmentStart(barrier.getId(), barrier.getTimestamp());
}
if (alignedChannels.size() == targetChannelCount) {
    // 所有通道到齐 → 结束对齐计时
    markAlignmentEnd();
}
...
if (alignedChannels.size() == targetChannelCount) {
    alignedChannels.clear();
    lastCancelledOrCompletedCheckpointId = currentCheckpointId;
    resetAlignmentTimer();
    allBarriersReceivedFuture.complete(null);
}

对齐耗时（第一个 Barrier 到达 → 全部到齐）和对齐期间处理的字节数写入 CheckpointMetrics，Web UI 的 Checkpoint 详情里可以看到。

3.3 对齐的核心：阻塞通道

对齐的核心在对齐状态机的 barrierReceived()：

1
2
3
4
5
6
7
8
9
10
11
12
13
// AbstractAlignedBarrierHandlerState.java
public final BarrierHandlerState barrierReceived(
        Controller controller, InputChannelInfo channelInfo,
        CheckpointBarrier checkpointBarrier, boolean markChannelBlocked)
        throws IOException, CheckpointException {
    if (markChannelBlocked) {
        state.blockChannel(channelInfo);    // ← 阻塞该通道
    }
    if (controller.allBarriersReceived()) {
        return triggerGlobalCheckpoint(controller, checkpointBarrier);   // ← 到齐，触发快照
    }
    return convertAfterBarrierReceived(state);
}

blockChannel() 是对齐的关键动作：

1
2
3
4
5
// ChannelState.java
public void blockChannel(InputChannelInfo channelInfo) {
    inputs[channelInfo.getGateIdx()].blockConsumption(channelInfo);
    blockedChannels.add(channelInfo);
}

“阻塞通道”阻塞的是什么？是 Barrier 之后数据的消费：

  • 该通道 Barrier 之前的数据已经处理（或正在处理）
  • Barrier 被消费后，Barrier 之后的数据留在网络层 buffer 中，不再被取走
  • 通道被”卡住”，等其他通道的 Barrier 到达

这就是对齐的本质——快通道被卡住等慢通道。假设两个通道，A 快 B 慢：

  1. A 的 Barrier 先到，A 被阻塞，A 后续的数据在 buffer 中堆积
  2. B 的 Barrier 到，B 也进入收集状态
  3. 两个通道到齐 → 触发快照 → 放行所有通道，堆积的数据一起处理

对齐的代价显而易见：快通道被阻塞期间数据在 buffer 堆积，形成反压，向上游传导，最终拖慢 Source 的读取速度。数据倾斜或某条边特别慢时，对齐时间会很长，甚至导致 Checkpoint 超时。这是非对齐 Checkpoint 的动机（第四节）。

3.4 对齐超时：alternating 切换非对齐

alternating 模式（对齐 + 超时）下，第一个 Barrier 到达时注册超时定时器：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
// SingleCheckpointBarrierHandler.java
private void registerAlignmentTimer(CheckpointBarrier announcedBarrier) {
    long alignedCheckpointTimeout =
            announcedBarrier.getCheckpointOptions().getAlignedCheckpointTimeout();
    long timePassedSinceCheckpointStart =
            getClock().absoluteTimeMillis() - announcedBarrier.getTimestamp();
    long timerDelay = Math.max(alignedCheckpointTimeout - timePassedSinceCheckpointStart, 0);

    this.currentAlignmentTimer = registerTimer.apply(
            () -> {
                // 超时后若 Barrier 还没收齐 → 切换非对齐
                if (currentCheckpointId == barrierId
                        && !getAllBarriersReceivedFuture(barrierId).isDone()) {
                    currentState = currentState.alignmentTimeout(context, announcedBarrier);
                }
                ...
            },
            Duration.ofMillis(timerDelay));
}

定时器注册在处理时间服务上，回调在 mailbox 线程执行。超时时间从 Barrier 的 timestamp（JobManager 触发时刻）算起，不是从第一个 Barrier 到达算起——因为 Barrier 在途传播的时间也算对齐成本的一部分。

超时后 alignmentTimeout() 做两件事：把已收到预告（announcement）的 Barrier 转为优先事件（prioritizeAllAnnouncements，让 Barrier 越过在途数据），然后切换到非对齐状态（见 4.2）。

为什么超时切非对齐而不是直接 abort？非对齐仍然能完成 Checkpoint（只是快照变大，多了在途数据），abort 则整轮失败。切非对齐是折中——用快照体积换 Checkpoint 的完成。

3.5 上游结束与取消

两个异常路径：

上游分区结束（processEndOfPartition）：上游通道结束（EndOfPartitionEvent）时，目标通道数减一。若当前有进行中的 Checkpoint：

  • 未开启”任务结束后仍可 Checkpoint”：直接 abort（该通道不会再有 Barrier，对齐不可能完成）
  • 已开启：把该通道视为”已对齐”，走状态机的 endOfPartitionReceived，可能触发快照

取消标记（processCancellationBarrier）：收到 CancelCheckpointMarker（上游 abort 后广播）时，abort 当前 Checkpoint：

1
2
3
4
5
6
7
8
9
10
// SingleCheckpointBarrierHandler.java
public void processCancellationBarrier(CancelCheckpointMarker cancelBarrier, InputChannelInfo channelInfo)
        throws IOException {
    final long cancelledId = cancelBarrier.getCheckpointId();
    if (cancelledId > currentCheckpointId
            || (cancelledId == currentCheckpointId && alignedChannels.size() > 0)) {
        abortInternal(cancelledId,
                new CheckpointException(CheckpointFailureReason.CHECKPOINT_DECLINED_ON_CANCELLATION_BARRIER));
    }
}

abortInternal() 做清理：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
// SingleCheckpointBarrierHandler.java
private void abortInternal(long cancelledId, CheckpointException exception) throws IOException {
    // 抬高 currentCheckpointId、清空 alignedChannels，后续旧 Barrier 无法再启动对齐
    currentCheckpointId = Math.max(cancelledId, currentCheckpointId);
    lastCancelledOrCompletedCheckpointId = Math.max(lastCancelledOrCompletedCheckpointId, cancelledId);
    pendingCheckpointBarrier = null;
    alignedChannels.clear();
    targetChannelCount = 0;
    resetAlignmentTimer();
    currentState = currentState.abort(cancelledId);    // 状态机 abort：放行所有被阻塞通道
    ...
    notifyAbort(cancelledId, exception);               // 通知 Task 侧 decline
    allBarriersReceivedFuture.completeExceptionally(exception);
}

notifyAbort 到 SubtaskCheckpointCoordinatorImpl.abortCheckpointOnBarrier()，做两件事：向 JobManager 上报 decline（触发 JM 端 abortPendingCheckpoint，见上一篇第三节），向下游广播 CancelCheckpointMarker（让下游也别等这个 Barrier）。

subsume：旧 Checkpoint 未完成时收到新 Checkpoint 的 Barrier，abort 旧的：

1
2
3
4
5
// SingleCheckpointBarrierHandler.java
private void cancelSubsumedCheckpoint(long barrierId) throws IOException {
    LOG.warn("{}: Received checkpoint barrier for checkpoint {} before completing current checkpoint {}. Skipping current checkpoint.", ...);
    abortInternal(currentCheckpointId, CHECKPOINT_DECLINED_SUBSUMED);
}

新 Barrier 取代旧 Checkpoint，下游不再等旧 Checkpoint 的 Barrier，避免无意义的对齐阻塞。

四、非对齐 Checkpoint

4.1 核心思想

对齐的代价是快通道被阻塞、反压上游。数据倾斜或某条边很慢时，对齐时间可能长到 Checkpoint 超时。

非对齐 Checkpoint 的核心思想：不等对齐，第一个 Barrier 到达就立即快照，把网络层里的”在途数据”（已收到未处理的数据）也一起快照下来。

也就是说，非对齐把 in-flight data 从网络层”搬”进 Checkpoint，快照不再依赖所有通道到齐。代价是快照体积变大（多了在途数据），恢复时要重放这些数据。

4.2 状态机：第一个 Barrier 立即触发

非对齐模式下状态机大幅简化：

1
2
3
4
5
6
7
8
AlternatingWaitingForFirstBarrierUnaligned（等待第一个 Barrier）
                    │
                    │ 第一个 Barrier
                    ▼
        立即触发快照（不阻塞、不对齐）
                    │
                    ├─ 所有通道到齐 → 结束（checkpointStopped，停止持久化）
                    └─ 未到齐 → AlternatingCollectingBarriersUnaligned（只计数剩余 Barrier）

关键在 AlternatingWaitingForFirstBarrierUnaligned.barrierReceived()：

1
2
3
4
5
6
7
8
9
10
// AlternatingWaitingForFirstBarrierUnaligned.java
public BarrierHandlerState barrierReceived(...) {
    CheckpointBarrier unalignedBarrier = checkpointBarrier.asUnaligned();
    controller.initInputsCheckpoint(unalignedBarrier);       // 1. 启动输入侧在途数据持久化
    for (CheckpointableInput input : channelState.getInputs()) {
        input.checkpointStarted(unalignedBarrier);           // 2. 网络层 spill 在途 buffer
    }
    controller.triggerGlobalCheckpoint(unalignedBarrier);    // 3. 立即触发快照
    ...
}

三步：

  1. initInputsCheckpoint：启动输入侧在途数据持久化
  2. checkpointStarted：网络层把在途 buffer 标记为持久化对象
  3. triggerGlobalCheckpoint：立即触发算子状态快照，不等其他通道

对比对齐模式：对齐的 barrierReceived 先 blockChannel 阻塞通道，所有通道到齐才触发快照；非对齐不阻塞，第一个 Barrier 直接触发。

4.3 输入侧：在途数据持久化

在途数据指已到达 Task 但还没处理、积压在网络层 buffer 中的数据。对齐模式下它不存在（通道被阻塞，Barrier 之后的数据进不了 Task）；非对齐模式下必须把它捕获下来。

initInputsCheckpoint() 启动捕获：

1
2
3
4
5
6
7
8
// SubtaskCheckpointCoordinatorImpl.java
public void initInputsCheckpoint(long id, CheckpointOptions checkpointOptions)
        throws CheckpointException {
    if (checkpointOptions.isUnalignedCheckpoint()) {
        channelStateWriter.start(id, checkpointOptions);
        prepareInflightDataSnapshot(id);    // 异步把在途数据落盘
    }
}

prepareInflightDataSnapshot() 异步读取各输入通道的在途 buffer，经 ChannelStateWriter 写入 Checkpoint 存储。网络层 RemoteInputChannel.checkpointStarted() 负责标记要持久化的 buffer：

1
2
3
4
5
6
7
8
// RemoteInputChannel.java
public void checkpointStarted(CheckpointBarrier barrier) throws CheckpointException {
    synchronized (receivedBuffers) {
        ...
        channelStatePersister.startPersisting(
                barrier.getId(), getInflightBuffersUnsafe(barrier.getId()));
    }
}

恢复时这些在途数据会被重新注入数据流，保证 exactly-once 语义不变。

4.4 输出侧

输出侧的 Barrier 作为优先事件广播（见 2.2 的 broadcastEvent 第二个参数）。优先事件越过在途输出数据，Barrier 之前已写入 buffer 的输出数据由 ChannelStateWriter 捕获持久化。

checkpointState() 的第 (3) 步标记输出侧捕获结束：

1
2
3
4
5
6
// SubtaskCheckpointCoordinatorImpl.java
// Step (3): Prepare to spill the in-flight buffers for input and output
if (options.isUnalignedCheckpoint()) {
    // output data already written while broadcasting event
    channelStateWriter.finishOutput(metadata.getCheckpointId());
}

4.5 限制

非对齐 Checkpoint 有几个限制：

  • Savepoint 不支持非对齐（CheckpointOptions.unaligned() 直接拒绝 Savepoint 类型）
  • 只适用于点对点拓扑。广播、all-to-all 等拓扑下在途数据量太大，会被强制转回对齐（FORCED_ALIGNED，checkpointState() 里有对应的还原逻辑）
  • 恢复/并行度变更时要重放在途数据，恢复成本高于对齐 Checkpoint

五、状态快照

对齐完成（或非对齐下第一个 Barrier 到达）后，triggerGlobalCheckpoint() → notifyCheckpoint() → StreamTask.triggerCheckpointOnBarrier() → performCheckpoint() → SubtaskCheckpointCoordinatorImpl.checkpointState() 执行快照。

5.1 同步阶段

checkpointState() 是快照主流程（同步阶段，mailbox 线程）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
// SubtaskCheckpointCoordinatorImpl.java
public void checkpointState(CheckpointMetaData metadata, CheckpointOptions options, ...) throws Exception {
    // Step (0): 记录 lastCheckpointId；若已被通知 abort，广播取消标记后返回
    lastCheckpointId = metadata.getCheckpointId();
    if (checkAndClearAbortedStatus(metadata.getCheckpointId())) {
        operatorChain.broadcastEvent(new CancelCheckpointMarker(metadata.getCheckpointId()));
        return;
    }

    // Step (1): 算子做 Barrier 前的准备工作
    operatorChain.prepareSnapshotPreBarrier(metadata.getCheckpointId());

    // Step (2): 把 Barrier 转发给下游（尽快发出）
    operatorChain.broadcastEvent(
            new CheckpointBarrier(metadata.getCheckpointId(), metadata.getTimestamp(), options),
            options.isUnalignedCheckpoint());

    // Step (3): 非对齐模式：结束输出侧在途数据捕获
    if (options.isUnalignedCheckpoint()) {
        channelStateWriter.finishOutput(metadata.getCheckpointId());
    }

    // Step (4): 状态快照（同步部分，拿 future）→ 提交异步阶段
    Map<OperatorID, OperatorSnapshotFutures> snapshotFutures = new HashMap<>();
    if (takeSnapshotSync(snapshotFutures, metadata, metrics, options, operatorChain, isRunning)) {
        finishAndReportAsync(snapshotFutures, metadata, metrics, ...);
    } else {
        cleanup(snapshotFutures, metadata, metrics, new Exception("Checkpoint declined"));
    }
}

步骤顺序有个设计问题：为什么先转发 Barrier（Step 2）再快照（Step 4）？

因为 Barrier 越早发出，下游越早开始对齐。本地快照只影响本 Task，Barrier 传播影响整条下游链路。先发出 Barrier 把对下游对齐的影响降到最低。源码注释也写明了这一点：”We generally try to emit the checkpoint barrier as soon as possible to not affect downstream checkpoint alignments”。

prepareSnapshotPreBarrier() 是算子的钩子，做 Barrier 前的准备工作，比如 AsyncWaitOperator 用它检查未完成的异步操作。这一步抛异常会 decline 本次 Checkpoint。

Step (0) 的 checkAndClearAbortedStatus 处理一个竞态：JM 的 abort 通知可能先于 Barrier 到达，此时不再执行快照，而是广播取消标记让下游别等。

5.2 单算子快照

takeSnapshotSync() → operatorChain.snapshotState() 遍历链上所有算子，每个算子走 StreamOperatorStateHandler.snapshotState()：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
// StreamOperatorStateHandler.java
void snapshotState(...) throws CheckpointException {
    try {
        // 1. 定时器快照（legacy 模式写入 raw keyed state）
        if (timeServiceManager.isPresent()) {
            ...
            manager.snapshotToRawKeyedState(...);
        }
        // 2. 用户算子的快照钩子（如 ListCheckpointed）
        streamOperator.snapshotState(snapshotContext);

        snapshotInProgress.setKeyedStateRawFuture(snapshotContext.getKeyedStateStreamFuture());
        snapshotInProgress.setOperatorStateRawFuture(snapshotContext.getOperatorStateStreamFuture());

        // 3. 算子状态（非 keyed）
        if (null != operatorStateBackend) {
            snapshotInProgress.setOperatorStateManagedFuture(
                    operatorStateBackend.snapshot(checkpointId, timestamp, factory, checkpointOptions));
        }

        // 4. keyed 状态（走 SnapshotStrategyRunner）
        if (null != keyedStateBackend) {
            snapshotInProgress.setKeyedStateManagedFuture(
                    keyedStateBackend.snapshot(checkpointId, timestamp, factory, checkpointOptions));
        }
    } catch (Exception snapshotException) {
        snapshotInProgress.cancel();
        ...
        throw new CheckpointException(..., CheckpointFailureReason.CHECKPOINT_DECLINED, snapshotException);
    }
}

四步，每步产生一个 future 存入 OperatorSnapshotFutures：

  • raw keyed state（定时器、自定义 raw state）
  • raw operator state
  • managed operator state（非 keyed 状态）
  • managed keyed state（keyed 状态，主体）

snapshotContext（StateSnapshotContextSynchronousImpl）提供快照的输出流，算子在 snapshotState() 里往里面写状态。

keyed 状态快照走 SnapshotStrategyRunner：

1
2
3
4
5
6
7
8
9
10
// SnapshotStrategyRunner.java
public final RunnableFuture<SnapshotResult<T>> snapshot(...) throws Exception {
    // 同步部分：准备资源（如 RocksDB 的 prepareSnapshotResources）
    SR snapshotResources = snapshotStrategy.syncPrepareResources(checkpointId);
    // 异步部分：真正写状态（返回 supplier，在异步阶段执行）
    SnapshotStrategy.SnapshotResultSupplier<T> asyncSnapshot =
            snapshotStrategy.asyncSnapshot(snapshotResources, checkpointId, timestamp,
                    streamFactory, checkpointOptions);
    ...
}

同步部分只做”准备”（RocksDB 是准备 SST 文件引用），真正的落盘在异步阶段完成。这就是同步阶段快、异步阶段慢的原因。

5.3 异步阶段与 ack

finishAndReportAsync() 把 AsyncCheckpointRunnable 提交到 asyncOperationsThreadPool：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
// AsyncCheckpointRunnable.java
public void run() {
    try {
        // 等待各算子快照 future 完成，组装 TaskStateSnapshot
        SnapshotsFinalizeResult snapshotsFinalizeResult = finalizeNonFinishedSnapshots();
        ...
        if (asyncCheckpointState.compareAndSet(RUNNING, COMPLETED)) {
            reportCompletedSnapshotStates(...);    // 上报 JobManager（ack）
        }
        finishedFuture.complete(null);
    } catch (Exception e) {
        handleExecutionException(e);               // 异步异常 → decline
        finishedFuture.completeExceptionally(e);
    } finally {
        unregisterConsumer.accept(this);
    }
}

finalizeNonFinishedSnapshots() 等待所有算子的快照 future 完成，组装两份 TaskStateSnapshot：

  • jobManagerTaskOperatorSubtaskStates：需要上报给 JM 的状态（managed state handle）
  • localTaskOperatorSubtaskStates：留在 TaskManager 本地的状态（raw state 等）

reportCompletedSnapshotStates() → taskStateManager.reportTaskStateSnapshots() 做两件事：

1
2
3
4
5
6
7
8
// TaskStateManagerImpl.java
public void reportTaskStateSnapshots(...) {
    // 1. 本地状态落盘到 TaskManager 本地目录
    localStateStore.storeLocalState(checkpointId, localState);
    // 2. RPC 向 JobManager 发 ack（携带 state handle）
    checkpointResponder.acknowledgeCheckpoint(
            jobId, executionAttemptID, checkpointId, checkpointMetrics, acknowledgedState);
}

这个 ack 就是上一篇 2.5 节 JobManager 收到的 AcknowledgeCheckpoint 消息。JM 收齐所有 Task 的 ack 后完成 Checkpoint。

5.4 完成通知

Checkpoint 完成后，JobManager 通知各 Task（notifyCheckpointComplete），Task 端：

1
2
3
4
5
6
7
8
9
10
11
12
// SubtaskCheckpointCoordinatorImpl.java
private void notifyCheckpoint(...) throws Exception {
    ...
    switch (notifyCheckpointOperation) {
        case COMPLETE:
            operatorChain.notifyCheckpointComplete(checkpointId);    // 通知算子
            break;
        ...
    }
    ...
    env.getTaskStateManager().notifyCheckpointComplete(checkpointId); // 释放本地状态
}

operatorChain.notifyCheckpointComplete() 通知链上每个算子，这是端到端 Exactly-Once 的关键：实现 TwoPhaseCommitSinkFunction 的 Sink（如 Kafka 事务）收到这个通知后才提交当前事务批次，数据真正可见。JM 端的通知机制见上一篇 2.6，这里只看 Task 端的接收路径。

六、Task 端失败处理

Task 端的 Checkpoint 失败主要有四类：

decline（主动拒绝）：

  • Task 不在运行态：performCheckpoint() 返回 false，Task.triggerCheckpointBarrier 上报 decline
  • prepareSnapshotPreBarrier() 抛异常：算子主动拒绝（如 AsyncWaitOperator 有未完成的异步操作）
  • 算子快照失败：StreamOperatorStateHandler.snapshotState() 抛 CHECKPOINT_DECLINED

abort（取消）：收到 CancelCheckpointMarker（上游 abort 后广播）→ abortInternal() → decline + 向下游广播。见 3.5。

异步异常：异步阶段失败 → AsyncCheckpointRunnable.handleExecutionException() → decline。

subsume：新 Barrier 到达时旧 Checkpoint 未完成 → abort 旧的。见 3.5。

所有 decline 都通过 env.declineCheckpoint() 发到 JobManager，JM 收到后调用 abortPendingCheckpoint()（上一篇第三节）。Task 端和 JM 端构成失败处理闭环：Task 上报 decline，JM abort 并通知其余 Task。

七、整体链路图

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
Source Task（RPC 触发）
  └─ StreamTask.triggerCheckpointAsync()
       └─ [mailbox] performCheckpoint()
            └─ SubtaskCheckpointCoordinatorImpl.checkpointState()
                 ├─ prepareSnapshotPreBarrier()
                 ├─ broadcastEvent(CheckpointBarrier)    ← Barrier 注入数据流
                 ├─ [非对齐] finishOutput()
                 └─ takeSnapshotSync() → 各算子快照（拿 future）
                      └─ finishAndReportAsync() → AsyncCheckpointRunnable [线程池]

中间 Task（Barrier 经网络到达）
  └─ CheckpointedInputGate.handleEvent()
       └─ SingleCheckpointBarrierHandler.processBarrier()
            ├─ checkNewCheckpoint()                  （subsume 旧 Checkpoint）
            ├─ [对齐] blockChannel()                 （阻塞快通道）
            ├─ [到齐] triggerGlobalCheckpoint()
            │    └─ StreamTask.triggerCheckpointOnBarrier()
            │         └─ performCheckpoint() → checkpointState()（同 Source 路径）
            ├─ [对齐超时] alignmentTimeout() → 切非对齐
            └─ [非对齐] 第一个 Barrier → initInputsCheckpoint + checkpointStarted
                 └─ triggerGlobalCheckpoint()（立即）

异步阶段（asyncOperationsThreadPool）
  └─ AsyncCheckpointRunnable.run()
       ├─ finalizeNonFinishedSnapshots()             （等 future，组装 TaskStateSnapshot）
       └─ reportTaskStateSnapshots()
            ├─ localStateStore.storeLocalState()     （本地状态落盘）
            └─ checkpointResponder.acknowledgeCheckpoint()  → [RPC] → JobManager

JobManager 收齐 ack → completePendingCheckpoint()（见上一篇）
  └─ notifyCheckpointOnComplete → Task
       └─ operatorChain.notifyCheckpointComplete()  （Sink 两阶段提交）

下一篇

Flink Checkpoint 源码（三）：StateBackend 源码分析，从 KeyedStateBackend.snapshot() 开始，分析 HashMapStateBackend 与 RocksDBStateBackend 的快照实现差异，以及 RocksDB 增量 Checkpoint 的 SST 文件管理。

赏

感谢您的支持!

支付宝
微信

本文由 董可伦 发表于 伦少的博客 ,采用署名-非商业性使用-禁止演绎 3.0进行许可。

非商业转载请注明作者及出处。商业转载请联系作者本人。

本文标题：Flink Checkpoint 源码（二）：Barrier 对齐与状态快照

本文链接：https://dongkelun.com/2026/10/13/flink/flink-checkpoint-barrier-alignment/

欢迎关注我的公众号

扫一扫，分享到微信

微信分享二维码

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

2016年本科毕业于合肥工业大学<br><br>信息与计算科学专业<br><br>目前在济南一家公司从事大数据的工作~~