Flink Checkpoint 源码（一）：CheckpointCoordinator 触发链路

2026-09-20

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

前言

本文是 Flink Checkpoint 源码分析的第一篇，聚焦 JobManager 端 CheckpointCoordinator 的触发链路：从定时调度、触发决策，到向各 Task 发送触发请求，再到收齐确认、完成 Checkpoint 的全过程。Barrier 的注入和传播属于 TaskManager 端的逻辑，将在下一篇展开。

前置知识：

Flink Checkpoint 原理：概念与整体流程

版本

Flink 1.15.3

一、CheckpointCoordinator 概览

CheckpointCoordinator 是 Checkpoint 的总调度者，运行在 JobManager 上，每个 Job 一个实例，承担三项职责：

触发：定时或按需发起 Checkpoint 请求
收集：接收各 Task 的快照确认（ack）
完成：收齐后持久化 Checkpoint 元数据，并通知 Sink 提交

它的主要组件：

CheckpointCoordinator
  ├── CheckpointRequestDecider      请求队列 + 决策（TreeSet，按优先级排序）
  ├── CheckpointPlanCalculator      计算本次 Checkpoint 涉及哪些 Task
  ├── CheckpointIDCounter           ID 分配（HA 模式走 ZooKeeper 保证单调递增）
  ├── pendingCheckpoints            进行中的 Checkpoint（LinkedHashMap<id, PendingCheckpoint>）
  ├── completedCheckpointStore      已完成的 Checkpoint（持久化元数据，淘汰旧的）
  ├── checkpointStorageView         Checkpoint 存储位置（目录初始化）
  ├── masterHooks                   外部系统钩子（如 HBase 协同快照）
  ├── failureManager                失败处理策略
  ├── timer                         单线程定时器（周期触发 + 超时取消）
  └── executor                      IO 线程池（存储初始化等异步任务）

两个线程池值得注意：timer 是单线程的，负责创建 PendingCheckpoint、abort、complete 等所有状态变更；executor 是多线程 IO 池，负责存储目录初始化等耗时操作。源码里很多方法都显式指定了运行线程，这是读触发链路的一条主线。

二、触发链路

一次 Checkpoint 的完整路径：

1	定时调度 → 请求入队 → 决策（执行 / 排队 / 丢弃）→ 异步准备 → 发送触发 RPC → 接收 ack → 完成

2.1 定时器何时启动

定时器不是在 CheckpointCoordinator 创建时启动的，而是等 Job 变为 RUNNING 才启动，分两阶段：

阶段一（注册）：JobMaster 构建 ExecutionGraph 时，如果 Checkpoint 启用，会创建 CheckpointCoordinator 并注册 Job 状态监听器：

JobMaster 构造方法
  └─ JobMaster.createScheduler()
       └─ SlotPoolServiceSchedulerFactory.createScheduler()
            └─ SchedulerNGFactory.createInstance()   （流作业：DefaultSchedulerFactory）
                 └─ new DefaultScheduler(...) → SchedulerBase 构造方法
                      └─ createAndRestoreExecutionGraph()
                           └─ DefaultExecutionGraphFactory.createAndRestoreExecutionGraph()
                                └─ DefaultExecutionGraphBuilder.buildGraph()
                                     └─ executionGraph.enableCheckpointing()   （仅 Checkpoint 启用时）
                                          ├─ new CheckpointCoordinator(...)
                                          └─ registerJobStatusListener(checkpointCoordinator.createActivatorDeactivator())

createActivatorDeactivator() 返回的 CheckpointCoordinatorDeActivator 实现了 JobStatusListener，存入 ExecutionGraph 的 jobStatusListeners 列表。

阶段二（回调）：Job 状态变化时，Scheduler 驱动状态机转换，ExecutionGraph 回调所有注册的监听器：

SchedulerBase.transitionExecutionGraphState(RUNNING)
  └─ DefaultExecutionGraph.transitionState()
       └─ notifyJobStatusChange(RUNNING)
            └─ CheckpointCoordinatorDeActivator.jobStatusChanges(RUNNING)
                 └─ coordinator.startCheckpointScheduler()

回调方法：

// CheckpointCoordinatorDeActivator.java
public void jobStatusChanges(JobID jobId, JobStatus newJobStatus, long timestamp) {
    if (newJobStatus == JobStatus.RUNNING) {
        coordinator.startCheckpointScheduler();
    } else {
        coordinator.stopCheckpointScheduler();
    }
}

Job 切到非 RUNNING 状态（比如恢复中）时，定时器停止，同时 abort 所有 pending 和排队中的 Checkpoint。

startCheckpointScheduler() 本身很短：

// CheckpointCoordinator.java

public void startCheckpointScheduler() {
    synchronized (lock) {
        Preconditions.checkState(
                isPeriodicCheckpointingConfigured(),
                "Can not start checkpoint scheduler, if no periodic checkpointing is configured");

        // 先确保取消旧的定时器
        stopCheckpointScheduler();

        periodicScheduling = true;
        currentPeriodicTrigger = scheduleTriggerWithDelay(getRandomInitDelay());
    }
}

// 在 [minPauseBetweenCheckpoints, baseInterval) 区间内随机取初始延迟值
// 目的是防止大规模集群中多个作业同时触发 Checkpoint 对存储后端造成压力尖峰
private long getRandomInitDelay() {
    return ThreadLocalRandom.current().nextLong(minPauseBetweenCheckpoints, baseInterval + 1L);
}

// 使用 scheduleAtFixedRate 固定速率调度，不受前一次 Checkpoint 执行时长影响
// 如果某次 Checkpoint 执行时间超过 baseInterval，下一次会立即触发（无间隔）
private ScheduledFuture<?> scheduleTriggerWithDelay(long initDelay) {
    return timer.scheduleAtFixedRate(
            new ScheduledTrigger(), initDelay, baseInterval, TimeUnit.MILLISECONDS);
}

// 定时任务 Runnable 实现。isPeriodic=true 标识周期性触发，异常仅记录日志不中断定时器
private final class ScheduledTrigger implements Runnable {
    @Override
    public void run() {
        try {
            triggerCheckpoint(true);
        } catch (Exception e) {
            LOG.error("Exception while triggering checkpoint for job {}.", job, e);
        }
    }
}

这里有两个设计点：

随机初始延迟：第一次触发不是立即执行，而是在 [minPauseBetweenCheckpoints, baseInterval] 内随机延迟。大集群里很多作业会同时启动（比如故障后批量重启），如果 Checkpoint 同时触发，存储后端（HDFS/S3）会迎来一波写尖峰。随机化初始延迟把触发时间打散。
scheduleAtFixedRate：后续触发间隔固定为 baseInterval（即 execution.checkpointing.interval），不受前一次 Checkpoint 耗时影响。某次 Checkpoint 耗时超过间隔时，下一次会在前一次结束后立即触发（是否真正执行由 2.2 的决策层判断）。备选方案 scheduleWithFixedDelay 是前一次结束后再等固定时间，实际间隔会被慢 Checkpoint 拉长。

ScheduledTrigger 吞掉异常只记日志，单次触发失败不会中断定时器。

2.2 多个请求如何决策

定时器、手动触发（REST API / CLI）、Savepoint 都走同一个入口：

// CheckpointCoordinator.java

// 统一入口方法，定时/手动/Savepoint 都走这里。三步：构造请求 → chooseRequestToExecute 决策 → 返回 onCompletionPromise
public CompletableFuture<CompletedCheckpoint> triggerCheckpoint(
        CheckpointProperties props,
        @Nullable String externalSavepointLocation,
        boolean isPeriodic) {

    CheckpointTriggerRequest request =
            new CheckpointTriggerRequest(props, externalSavepointLocation, isPeriodic);
    chooseRequestToExecute(request).ifPresent(this::startTriggeringCheckpoint);
    return request.onCompletionPromise;
}

请求先进入 CheckpointRequestDecider 内部的 NavigableSet（TreeSet），排序优先级：

Savepoint 优先于 Checkpoint
Force 优先于非 Force
手动触发（非周期性）优先于周期性
按时间戳先到先得

队列容量 1000，超限时丢弃优先级最低的那个——周期性请求先被牺牲，用户提交的请求尽量保留。

入队后由 chooseRequestToExecute() 决定是否执行：

// CheckpointRequestDecider.java

private Optional<CheckpointTriggerRequest> chooseRequestToExecute(
        boolean isTriggering, long lastCompletionMs) {

    // 条件1：正在触发中 / 队列为空 / 清理中的 Checkpoint 超过上限 → 不执行
    if (isTriggering
            || queuedRequests.isEmpty()
            || numberOfCleaningCheckpointsSupplier.getAsInt()
                    > maxConcurrentCheckpointAttempts) {
        return Optional.empty();
    }

    // 条件2：进行中的 Checkpoint 达到上限 → 只有 force 请求可执行
    if (pendingCheckpointsSizeSupplier.getAsInt() >= maxConcurrentCheckpointAttempts) {
        return Optional.of(queuedRequests.first())
                .filter(CheckpointTriggerRequest::isForce)
                .map(unused -> queuedRequests.pollFirst());
    }

    // 条件3：非 force 的周期性请求 → 检查距上次完成的最小间隔
    CheckpointTriggerRequest first = queuedRequests.first();
    if (!first.isForce() && first.isPeriodic) {
        long nextTriggerDelayMillis = nextTriggerDelayMillis(lastCompletionMs);
        if (nextTriggerDelayMillis > 0) {
            queuedRequests.pollFirst()
                    .completeExceptionally(
                            new CheckpointException(MINIMUM_TIME_BETWEEN_CHECKPOINTS));
            rescheduleTrigger.accept(nextTriggerDelayMillis);
            return Optional.empty();
        }
    }

    return Optional.of(queuedRequests.pollFirst());
}

决策结果分三种：

情况	结果
`isTriggering` 为 true，或清理中的 Checkpoint 超上限	不执行，请求留在队列等下次决策
进行中的 Checkpoint 达到上限（默认 1）	只有 force 请求（Savepoint / 手动强制）可执行，其余等待
非 force 周期性请求，距上次完成不足 `minPauseBetweenCheckpoints`	丢弃本次请求，重新调度定时器在剩余时间后再触发

第三种情况最容易误解：最小间隔不满足时，周期性请求是丢弃而不是排队。原因是周期性请求”可再生”——定时器下一轮还会触发，没必要占着队列。丢弃也避免了慢 Checkpoint 场景下的请求堆积（Checkpoint 耗时超过间隔时定时器会持续触发，若排队则队列不断增长）。同时 rescheduleTrigger 会取消原固定速率定时器，改调单次延迟触发，让下次触发时间对齐真实的间隔要求。

isTriggering 标志位在 startTriggeringCheckpoint 开始时置 true，触发 RPC 发出后（onTriggerSuccess，失败时走 onTriggerFailure）置回 false，只覆盖异步准备阶段。”任意时刻最多一个进行中的 Checkpoint”不是靠这个标志位保证的，而是靠条件 2 的 pending 数量检查。

2.3 触发前的异步准备

决策通过后进入 startTriggeringCheckpoint()。在向 Task 发触发请求之前，要先完成一系列准备：计算 Task 列表、分配 ID、创建存储目录、给 OperatorCoordinator 和外部系统做快照。这些步骤用 CompletableFuture 串成一条链：

calculateCheckpointPlan()                        [executor]
  → getAndIncrement() + createPendingCheckpoint()  [executor → timer]
    → initializeCheckpointLocation()               [executor]
      → OperatorCoordinator 快照                    [timer]
        → snapshotMasterState()                    [timer]
          → triggerCheckpointRequest()             [timer]

// CheckpointCoordinator.java

// 异步准备方法。6 步：计算 CheckpointPlan → 分配 checkpointID + 创建 PendingCheckpoint →
// 初始化 CheckpointStorageLocation → OperatorCoordinator Checkpoints → Master Hooks 快照 →
// 全部完成后调 triggerCheckpointRequest
private void startTriggeringCheckpoint(CheckpointTriggerRequest request) {
    try {
        synchronized (lock) {
            preCheckGlobalState(request.isPeriodic);
        }

        Preconditions.checkState(!isTriggering);
        isTriggering = true;

        final long timestamp = System.currentTimeMillis();

        // 第 1 步：计算 CheckpointPlan（executor 线程）
        CompletableFuture<CheckpointPlan> checkpointPlanFuture =
                checkpointPlanCalculator.calculateCheckpointPlan();

        // 第 2 步：生成 checkpointID + 创建 PendingCheckpoint（executor → timer 线程）
        final CompletableFuture<PendingCheckpoint> pendingCheckpointCompletableFuture =
                checkpointPlanFuture
                        .thenApplyAsync(plan -> {
                            long checkpointID = checkpointIdCounter.getAndIncrement();
                            return new Tuple2<>(plan, checkpointID);
                        }, executor)
                        .thenApplyAsync(checkpointInfo ->
                                createPendingCheckpoint(timestamp, request.props,
                                        checkpointInfo.f0, request.isPeriodic,
                                        checkpointInfo.f1, request.getOnCompletionFuture()),
                                timer);

        // 第 3 步：初始化 CheckpointStorageLocation（executor 线程）
        final CompletableFuture<?> coordinatorCheckpointsComplete =
                pendingCheckpointCompletableFuture
                        .thenApplyAsync(pendingCheckpoint -> {
                            CheckpointStorageLocation location = initializeCheckpointLocation(
                                    pendingCheckpoint.getCheckpointID(), request.props,
                                    request.externalSavepointLocation, initializeBaseLocations);
                            return Tuple2.of(pendingCheckpoint, location);
                        }, executor)
                        .thenComposeAsync(checkpointInfo -> {
                            // 第 4 步：OperatorCoordinator Checkpoints（timer 线程）
                            return OperatorCoordinatorCheckpoints
                                    .triggerAndAcknowledgeAllCoordinatorCheckpointsWithCompletion(
                                            coordinatorsToCheckpoint, checkpointInfo.f0, timer);
                        }, timer);

        // 第 5 步：Master Hooks 快照（timer 线程）
        final CompletableFuture<?> masterStatesComplete =
                coordinatorCheckpointsComplete.thenComposeAsync(ignored -> {
                    PendingCheckpoint checkpoint = FutureUtils.getWithoutException(
                            pendingCheckpointCompletableFuture);
                    return snapshotMasterState(checkpoint);
                }, timer);

        // 第 6 步：全部准备完成 → triggerCheckpointRequest()
        FutureUtils.assertNoException(
                CompletableFuture.allOf(masterStatesComplete, coordinatorCheckpointsComplete)
                        .handleAsync((ignored, throwable) -> {
                            final PendingCheckpoint checkpoint =
                                    FutureUtils.getWithoutException(
                                            pendingCheckpointCompletableFuture);
                            if (throwable != null) {
                                onTriggerFailure(checkpoint, throwable);
                            } else {
                                triggerCheckpointRequest(request, timestamp, checkpoint);
                            }
                            return null;
                        }, timer));
    } catch (Throwable throwable) {
        onTriggerFailure(request, throwable);
    }
}

几个理解要点：

线程分配：耗时 IO（存储目录初始化）跑在 executor 线程池，状态变更（创建 PendingCheckpoint、coordinator 快照）跑在单线程 timer 上，保证所有状态变更串行化，不用考虑并发。
createPendingCheckpoint 里还埋了一个超时：PendingCheckpoint 创建后会在 timer 上调度一个 CheckpointCanceller，超过 checkpointTimeout 未完成就 abort（见第三节）。
顺序约束：OperatorCoordinator 快照必须先于 Task 触发完成（支持 ExternallyInducedSource，需要先拿到 coordinator 状态再触发 Task），Master Hooks 又必须在 OperatorCoordinator 之后。
失败：任何一步失败都走 onTriggerFailure()（见第三节）。

2.4 如何触发 Task

准备完成后进入 triggerCheckpointRequest()，按顺序做三件事：

// CheckpointCoordinator.java

// 异步准备完成后调用，负责收尾。三步：triggerTasks（发 RPC）→ afterSourceBarrierInjection（打开事件阀门）→ maybeCompleteCheckpoint（尝试提前完成）
private void triggerCheckpointRequest(
        CheckpointTriggerRequest request, long timestamp, PendingCheckpoint checkpoint) {
    // 双重检查：异步准备期间 PendingCheckpoint 可能已被 abort
    if (checkpoint.isDisposed()) {
        onTriggerFailure(
                checkpoint,
                new CheckpointException(
                        CheckpointFailureReason.TRIGGER_CHECKPOINT_FAILURE,
                        checkpoint.getFailureCause()));
    } else {
        // 第 1 步：向所有需要触发的 Task 发送触发 RPC
        triggerTasks(request, timestamp, checkpoint)
                .exceptionally(failure -> {
                    // RPC 发送失败 → 包装为 CheckpointException
                    final CheckpointException cause = ...;

                    // 必须在 timer 线程内加锁 abort，保证线程安全
                    timer.execute(() -> {
                        synchronized (lock) {
                            abortPendingCheckpoint(checkpoint, cause);
                        }
                    });
                    return null;
                });

        // 第 2 步：打开 OperatorCoordinator 的事件阀门
        coordinatorsToCheckpoint.forEach(
                (ctx) -> ctx.afterSourceBarrierInjection(checkpoint.getCheckpointID()));

        // 第 3 步：尝试提前完成
        if (maybeCompleteCheckpoint(checkpoint)) {
            onTriggerSuccess();
        }
    }
}

第 1 步 triggerTasks 遍历 CheckpointPlan 里的 Task 列表，逐个发 RPC：

// CheckpointCoordinator.java

// 遍历 getTasksToTrigger() 逐个调用 Execution.triggerCheckpoint() 发 RPC，返回 waitForAll 聚合 Future
private CompletableFuture<Void> triggerTasks(
        CheckpointTriggerRequest request, long timestamp, PendingCheckpoint checkpoint) {
    final long checkpointId = checkpoint.getCheckpointID();

    final SnapshotType type;
    if (this.forceFullSnapshot && !request.props.isSavepoint()) {
        type = CheckpointType.FULL_CHECKPOINT;
    } else {
        type = request.props.getCheckpointType();
    }

    // Barrier 传递选项：全量/增量、exactly-once、是否允许非对齐、对齐超时
    final CheckpointOptions checkpointOptions = CheckpointOptions.forConfig(
            type,
            checkpoint.getCheckpointStorageLocation().getLocationReference(),
            isExactlyOnceMode,
            unalignedCheckpointsEnabled,
            alignedCheckpointTimeout);

    List<CompletableFuture<Acknowledge>> acks = new ArrayList<>();
    for (Execution execution : checkpoint.getCheckpointPlan().getTasksToTrigger()) {
        if (request.props.isSynchronous()) {
            // 同步 Savepoint：等 Task 完成快照再返回
            acks.add(execution.triggerSynchronousSavepoint(checkpointId, timestamp, checkpointOptions));
        } else {
            // 异步 Checkpoint：RPC 发出后立即返回，Task 完成后异步发 ack
            acks.add(execution.triggerCheckpoint(checkpointId, timestamp, checkpointOptions));
        }
    }
    return FutureUtils.waitForAll(acks);
}

这里的 RPC 只是告诉 Task”请执行 checkpointId 的快照”，并不包含 Barrier 本身，Barrier 由 TaskManager 端注入数据流（下一篇展开）。CheckpointOptions 携带 Barrier 传递策略（exactly-once、是否允许非对齐、对齐超时），TaskManager 端据此选择对齐策略。

第 2 步 afterSourceBarrierInjection 打开 OperatorCoordinator 的事件阀门。这个名字容易误解——“SourceBarrierInjection”指的是 CheckpointCoordinator 发出 triggerTasks RPC 这个动作，而不是 Barrier 真正插入数据流。此时 RPC 刚发出，Barrier 还没到 Task。

阀门机制是这样的：Coordinator 自身状态快照完成后，它的 OperatorEventValve 被关闭，之后通过 SubtaskGateway.sendEvent() 发出的 OperatorEvent（Split 分配、动态分区发现等）会被缓存到列表里，不发给 Task。这些事件是基于 Coordinator 快照之后的状态产生的，属于”下一个 epoch”，如果立即发出，可能抢在 Barrier 之前到达 Task（OperatorEvent 和 Checkpoint Barrier 走的是不同 RPC 通道），Task 处理 Barrier 时的状态就和 Barrier 之前不一致了。触发 RPC 发出后打开阀门，缓存的事件逐个发出，保证控制事件在 Barrier 之后才生效。

第 3 步 maybeCompleteCheckpoint 是防御性检查：极端场景下（比如 Task 与 JobManager 同进程、网络延迟极低），所有 ack 可能在 triggerTasks 返回前就到齐了，此时可以直接完成，不用等 ack 回调。

2.5 如何接收 ack

各 Task 完成快照后，把 AcknowledgeCheckpoint 消息发回 JobManager：

TaskManager 侧：
  RuntimeEnvironment.acknowledgeCheckpoint()
    → RpcCheckpointResponder.acknowledgeCheckpoint()
      → [RPC] CheckpointCoordinatorGateway.acknowledgeCheckpoint()

JobManager 侧：
  JobMaster.acknowledgeCheckpoint()          ← RPC 入口
    → SchedulerBase.acknowledgeCheckpoint()
      → ExecutionGraphHandler.acknowledgeCheckpoint()
        → CheckpointCoordinator.receiveAcknowledgeMessage()

receiveAcknowledgeMessage 的核心逻辑：

// CheckpointCoordinator.java

// 收到 Task 的 AcknowledgeCheckpoint 消息后的处理。注册共享状态、调用 acknowledgeTask、检查 isFullyAcknowledged
public boolean receiveAcknowledgeMessage(
        AcknowledgeCheckpoint message, String taskManagerLocationInfo)
        throws CheckpointException {
    ...
    synchronized (lock) {
        final PendingCheckpoint checkpoint = pendingCheckpoints.get(checkpointId);

        // 注册共享状态（无论 Checkpoint 是否还在进行中）
        if (message.getSubtaskState() != null
                && (checkpoint == null || !checkpoint.getProps().isSavepoint())) {
            message.getSubtaskState().registerSharedStates(
                    completedCheckpointStore.getSharedStateRegistry(), checkpointId);
        }

        if (checkpoint != null && !checkpoint.isDisposed()) {
            switch (checkpoint.acknowledgeTask(
                    message.getTaskExecutionId(), message.getSubtaskState(),
                    message.getCheckpointMetrics())) {
                case SUCCESS:
                    if (checkpoint.isFullyAcknowledged()) {
                        completePendingCheckpoint(checkpoint);
                    }
                    break;
                case DUPLICATE:
                    // 重复确认，忽略
                    break;
                case UNKNOWN:
                case DISCARDED:
                    // 丢弃 state handle，避免状态残留
                    discardSubtaskState(...);
            }
            return true;
        }
        // Checkpoint 已不存在（消息来晚了）→ 丢弃 state
        ...
    }
}

acknowledgeTask 把这个 Task 的快照状态（state handle）记入 PendingCheckpoint，返回四种结果之一：

结果	含义	处理
SUCCESS	正常确认	检查是否所有 Task 都已 ack，是则完成
DUPLICATE	同一 Task 重复确认	忽略
UNKNOWN	Task 的 execution attempt 未知（Task 已重启）	丢弃 state handle，避免状态残留
DISCARDED	PendingCheckpoint 已被 abort	丢弃 state handle

共享状态（多个 Task 共用的 state）会先注册进 SharedStateRegistry，它的生命周期跟随 Checkpoint 的 subsume（淘汰）而不是 ack，所以即使消息迟到也不会丢。

2.6 如何完成

所有 Task ack 收齐后调用 completePendingCheckpoint()（必须在锁内调用）：

// CheckpointCoordinator.java

// 所有 Task 确认后调用。finalizeCheckpoint → 存入 CompletedCheckpointStore → reportCompletedCheckpoint → scheduleTriggerRequest → cleanupAfterCompletedCheckpoint
private void completePendingCheckpoint(PendingCheckpoint pendingCheckpoint)
        throws CheckpointException {
    final long checkpointId = pendingCheckpoint.getCheckpointID();
    final CompletedCheckpoint completedCheckpoint;
    final CompletedCheckpoint lastSubsumed;
    final CheckpointProperties props = pendingCheckpoint.getProps();

    completedCheckpointStore.getSharedStateRegistry().checkpointCompleted(checkpointId);

    try {
        // 1. 收尾：PendingCheckpoint → CompletedCheckpoint（含所有 Task 的 state handle）
        completedCheckpoint = finalizeCheckpoint(pendingCheckpoint);

        // 2. 存入 store，超出保留数量则淘汰最旧的并清理
        if (!props.isSavepoint()) {
            lastSubsumed = addCompletedCheckpointToStoreAndSubsumeOldest(
                    checkpointId, completedCheckpoint,
                    pendingCheckpoint.getCheckpointPlan().getTasksToCommitTo());
        } else {
            lastSubsumed = null;
        }

        // 3. 上报统计
        reportCompletedCheckpoint(completedCheckpoint);
    } finally {
        // 4. 从进行中集合移除，执行队列中的下一个请求
        pendingCheckpoints.remove(checkpointId);
        scheduleTriggerRequest();
    }

    // 5. 清理与通知
    cleanupAfterCompletedCheckpoint(
            pendingCheckpoint, checkpointId, completedCheckpoint, lastSubsumed, props);
}

cleanupAfterCompletedCheckpoint 做两件重要的事：

// CheckpointCoordinator.java

private void cleanupAfterCompletedCheckpoint(
        PendingCheckpoint pendingCheckpoint,
        long checkpointId,
        CompletedCheckpoint completedCheckpoint,
        CompletedCheckpoint lastSubsumed,
        CheckpointProperties props) {
    rememberRecentCheckpointId(checkpointId);

    // 记录完成时间，用于 minPauseBetweenCheckpoints 的决策
    lastCheckpointCompletionRelativeTime = clock.relativeTimeMillis();

    logCheckpointInfo(completedCheckpoint);

    if (!props.isSavepoint() || props.isSynchronous()) {
        // 丢弃被 subsume 的 pending checkpoints
        dropSubsumedCheckpoints(checkpointId);

        // 通知各 Task 和 OperatorCoordinator Checkpoint 完成
        sendAcknowledgeMessages(
                pendingCheckpoint.getCheckpointPlan().getTasksToCommitTo(),
                checkpointId,
                completedCheckpoint.getTimestamp(),
                extractIdIfDiscardedOnSubsumed(lastSubsumed));
    }
}

void sendAcknowledgeMessages(
        List<ExecutionVertex> tasksToCommit,
        long completedCheckpointId,
        long completedTimestamp,
        long lastSubsumedCheckpointId) {
    // 通知 Task
    for (ExecutionVertex ev : tasksToCommit) {
        Execution ee = ev.getCurrentExecutionAttempt();
        if (ee != null) {
            ee.notifyCheckpointOnComplete(
                    completedCheckpointId, completedTimestamp, lastSubsumedCheckpointId);
        }
    }
    // 通知 OperatorCoordinator
    for (OperatorCoordinatorCheckpointContext coordinatorContext : coordinatorsToCheckpoint) {
        coordinatorContext.notifyCheckpointComplete(completedCheckpointId);
    }
}

notifyCheckpointOnComplete 通知是端到端 Exactly-Once 的关键：实现了 TwoPhaseCommitSinkFunction 的 Sink（如 Kafka 事务）收到这个通知后，才提交当前事务批次，数据真正可见。Checkpoint 失败被 abort 时，Sink 收到的是 notifyCheckpointAborted，事务回滚。

finally 块里的 scheduleTriggerRequest() 在 timer 线程上调度 executeQueuedRequest，重新走一遍 2.2 的决策逻辑，如果队列里有等待的请求（比如之前排队的手动触发），立即执行。这是”完成一个、接着执行下一个”的闭环。

三、失败处理

Checkpoint 失败分两个阶段，入口不同：

准备阶段失败（startTriggeringCheckpoint 的 Future 链中任何一步失败）走 onTriggerFailure()：

// CheckpointCoordinator.java

// 异步链路中任何一步失败的处理：abort OperatorCoordinator → abort PendingCheckpoint → failureManager 处理 → 重置 isTriggering → 执行队列下一个
private void onTriggerFailure(
        @Nullable PendingCheckpoint checkpoint,
        CheckpointProperties checkpointProperties,
        Throwable throwable) {
    throwable = ExceptionUtils.stripCompletionException(throwable);

    try {
        // 1. abort OperatorCoordinator 的触发
        coordinatorsToCheckpoint.forEach(
                OperatorCoordinatorCheckpointContext::abortCurrentTriggering);

        final CheckpointException cause = getCheckpointException(
                CheckpointFailureReason.TRIGGER_CHECKPOINT_FAILURE, throwable);

        if (checkpoint != null && !checkpoint.isDisposed()) {
            // 2. PendingCheckpoint 已创建 → abort 它（内部会调 failureManager）
            synchronized (lock) {
                abortPendingCheckpoint(checkpoint, cause);
            }
        } else {
            // 3. PendingCheckpoint 还没创建 → 直接交给 failureManager
            failureManager.handleCheckpointException(
                    checkpoint, checkpointProperties, cause, null, job, null, statsTracker);
        }
    } finally {
        // 4. 重置触发标志，执行队列中的下一个请求
        isTriggering = false;
        executeQueuedRequest();
    }
}

执行阶段失败（Task decline、超时、RPC 失败）走 abortPendingCheckpoint()：

// CheckpointCoordinator.java

private void abortPendingCheckpoint(
        PendingCheckpoint pendingCheckpoint,
        CheckpointException exception,
        @Nullable final ExecutionAttemptID executionAttemptID) {

    if (!pendingCheckpoint.isDisposed()) {
        try {
            // 释放资源：丢弃 state、清理存储
            pendingCheckpoint.abort(
                    exception.getCheckpointFailureReason(),
                    exception.getCause(),
                    checkpointsCleaner,
                    this::scheduleTriggerRequest,
                    executor,
                    statsTracker);

            failureManager.handleCheckpointException(
                    pendingCheckpoint,
                    pendingCheckpoint.getProps(),
                    exception,
                    executionAttemptID,
                    job,
                    getStatsCallback(pendingCheckpoint),
                    statsTracker);
        } finally {
            // 通知 Task abort（Sink 回滚事务）
            sendAbortedMessages(
                    pendingCheckpoint.getCheckpointPlan().getTasksToCommitTo(),
                    pendingCheckpoint.getCheckpointId(),
                    pendingCheckpoint.getCheckpointTimestamp());
            pendingCheckpoints.remove(pendingCheckpoint.getCheckpointId());
            rememberRecentCheckpointId(pendingCheckpoint.getCheckpointId());
            scheduleTriggerRequest();
        }
    }
}

执行阶段的失败来源主要有三种：

Task decline：Task 主动上报无法执行快照（如算子正在关闭、对齐限制超出），发送 DeclineCheckpoint 消息，JobManager 收到后调用 abortPendingCheckpoint
超时：createPendingCheckpoint 时在 timer 上调度了 CheckpointCanceller，超过 checkpointTimeout 未完成就 abort
触发 RPC 失败：triggerTasks 返回的 Future 异常时 abort

最终的失败策略由 CheckpointFailureManager 决定，分两个层面：

Job 级失败（准备阶段失败）：周期性 Checkpoint 失败计入连续失败计数器，超过 execution.checkpointing.tolerable-failed-checkpoints（默认无限）才 fail Job；Savepoint 失败不计入，异常通过 onCompletionPromise 抛给调用方
Task 级失败（执行阶段失败）：同步 Savepoint 直接 fail Job（避免 cancel 等待永远完不成的 Savepoint 造成死锁）；其他失败计入连续失败计数器，超限时 fail 对应 Task 触发 failover

只有特定失败原因会计入计数器（IO 异常、异步异常、decline、超时、finalize 失败等），minPause 不满足这类”非真失败”不计入。

四、整体链路图

Job RUNNING
  └─ CheckpointCoordinatorDeActivator.jobStatusChanges()
       └─ startCheckpointScheduler()
            └─ timer.scheduleAtFixedRate(ScheduledTrigger, 随机初始延迟, baseInterval)
                 └─ triggerCheckpoint(true)
                      └─ CheckpointRequestDecider 决策
                           │  执行
                           ▼
                      startTriggeringCheckpoint()
                           ├─ calculateCheckpointPlan()                  [executor]
                           ├─ getAndIncrement() + createPendingCheckpoint [executor→timer]
                           ├─ initializeCheckpointLocation()             [executor]
                           ├─ OperatorCoordinator 快照                    [timer]
                           ├─ snapshotMasterState()                      [timer]
                           ▼
                      triggerCheckpointRequest()
                           ├─ triggerTasks() → 向各 Task 发触发 RPC
                           ├─ afterSourceBarrierInjection() → 打开事件阀门
                           └─ maybeCompleteCheckpoint() → 尝试提前完成
                           │
                           │  （Barrier 传播与快照，下一篇展开）
                           ▼
                      receiveAcknowledgeMessage() × N
                           └─ acknowledgeTask() → isFullyAcknowledged?
                                │  否 → 等待更多 ack
                                │  是
                                ▼
                           completePendingCheckpoint()
                                ├─ finalizeCheckpoint → CompletedCheckpoint
                                ├─ CompletedCheckpointStore（超保留数淘汰旧的）
                                ├─ notifyCheckpointOnComplete → Sink 事务提交
                                └─ scheduleTriggerRequest → 执行队列下一个请求

Flink Checkpoint 源码（二）：Barrier 对齐与状态快照，从 StreamTask.performCheckpoint() 开始，分析 Barrier 在算子间的传播以及对齐/非对齐策略的源码实现。

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

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

本文标题：Flink Checkpoint 源码（一）：CheckpointCoordinator 触发链路

本文链接：https://dongkelun.com/2026/09/20/flink/flink-checkpoint-coordinator/

欢迎关注我的公众号