Flink Checkpoint 源码(一):CheckpointCoordinator 触发链路

前言

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

前置知识：

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

版本

Flink 1.15.3

一、CheckpointCoordinator 概览

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

1. 1. 触发：定时或按需发起 Checkpoint 请求
2. 2. 收集：接收各 Task 的快照确认（ack）
3. 3. 完成：收齐后持久化 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 的完整路径：

定时调度 → 请求入队 → 决策（执行 / 排队 / 丢弃）→ 异步准备 → 发送触发 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.javapublicvoidjobStatusChanges(JobID jobId, JobStatus newJobStatus, long timestamp) {if (newJobStatus == JobStatus.RUNNING) {        coordinator.startCheckpointScheduler();    } else {        coordinator.stopCheckpointScheduler();    }}

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

startCheckpointScheduler() 本身很短：

// CheckpointCoordinator.javapublicvoidstartCheckpointScheduler() {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 对存储后端造成压力尖峰privatelonggetRandomInitDelay() {return ThreadLocalRandom.current().nextLong(minPauseBetweenCheckpoints, baseInterval + 1L);}// 使用 scheduleAtFixedRate 固定速率调度，不受前一次 Checkpoint 执行时长影响// 如果某次 Checkpoint 执行时间超过 baseInterval，下一次会立即触发（无间隔）private ScheduledFuture<?> scheduleTriggerWithDelay(long initDelay) {return timer.scheduleAtFixedRate(newScheduledTrigger(), initDelay, baseInterval, TimeUnit.MILLISECONDS);}// 定时任务 Runnable 实现。isPeriodic=true 标识周期性触发，异常仅记录日志不中断定时器privatefinalclassScheduledTriggerimplementsRunnable {@Overridepublicvoidrun() {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 决策 → 返回 onCompletionPromisepublic CompletableFuture<CompletedCheckpoint> triggerCheckpoint(        CheckpointProperties props,@Nullable String externalSavepointLocation,boolean isPeriodic) {CheckpointTriggerRequestrequest=newCheckpointTriggerRequest(props, externalSavepointLocation, isPeriodic);    chooseRequestToExecute(request).ifPresent(this::startTriggeringCheckpoint);return request.onCompletionPromise;}

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

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

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

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

// CheckpointRequestDecider.javaprivate 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 的周期性请求 → 检查距上次完成的最小间隔CheckpointTriggerRequestfirst= queuedRequests.first();if (!first.isForce() && first.isPeriodic) {longnextTriggerDelayMillis= nextTriggerDelayMillis(lastCompletionMs);if (nextTriggerDelayMillis > 0) {            queuedRequests.pollFirst()                    .completeExceptionally(newCheckpointException(MINIMUM_TIME_BETWEEN_CHECKPOINTS));            rescheduleTrigger.accept(nextTriggerDelayMillis);return Optional.empty();        }    }return Optional.of(queuedRequests.pollFirst());}

决策结果分三种：

第三种情况最容易误解：最小间隔不满足时，周期性请求是丢弃而不是排队。原因是周期性请求"可再生"——定时器下一轮还会触发，没必要占着队列。丢弃也避免了慢 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 快照 →// 全部完成后调 triggerCheckpointRequestprivatevoidstartTriggeringCheckpoint(CheckpointTriggerRequest request) {try {synchronized (lock) {            preCheckGlobalState(request.isPeriodic);        }        Preconditions.checkState(!isTriggering);        isTriggering = true;finallongtimestamp= System.currentTimeMillis();// 第 1 步：计算 CheckpointPlan（executor 线程）        CompletableFuture<CheckpointPlan> checkpointPlanFuture =                checkpointPlanCalculator.calculateCheckpointPlan();// 第 2 步：生成 checkpointID + 创建 PendingCheckpoint（executor → timer 线程）final CompletableFuture<PendingCheckpoint> pendingCheckpointCompletableFuture =                checkpointPlanFuture                        .thenApplyAsync(plan -> {longcheckpointID= checkpointIdCounter.getAndIncrement();returnnewTuple2<>(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 -> {CheckpointStorageLocationlocation= 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 -> {PendingCheckpointcheckpoint= FutureUtils.getWithoutException(                            pendingCheckpointCompletableFuture);return snapshotMasterState(checkpoint);                }, timer);// 第 6 步：全部准备完成 → triggerCheckpointRequest()        FutureUtils.assertNoException(                CompletableFuture.allOf(masterStatesComplete, coordinatorCheckpointsComplete)                        .handleAsync((ignored, throwable) -> {finalPendingCheckpointcheckpoint=                                    FutureUtils.getWithoutException(                                            pendingCheckpointCompletableFuture);if (throwable != null) {                                onTriggerFailure(checkpoint, throwable);                            } else {                                triggerCheckpointRequest(request, timestamp, checkpoint);                            }returnnull;                        }, 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（尝试提前完成）privatevoidtriggerCheckpointRequest(        CheckpointTriggerRequest request, long timestamp, PendingCheckpoint checkpoint) {// 双重检查：异步准备期间 PendingCheckpoint 可能已被 abortif (checkpoint.isDisposed()) {        onTriggerFailure(                checkpoint,newCheckpointException(                        CheckpointFailureReason.TRIGGER_CHECKPOINT_FAILURE,                        checkpoint.getFailureCause()));    } else {// 第 1 步：向所有需要触发的 Task 发送触发 RPC        triggerTasks(request, timestamp, checkpoint)                .exceptionally(failure -> {// RPC 发送失败 → 包装为 CheckpointExceptionfinalCheckpointExceptioncause= ...;// 必须在 timer 线程内加锁 abort，保证线程安全                    timer.execute(() -> {synchronized (lock) {                            abortPendingCheckpoint(checkpoint, cause);                        }                    });returnnull;                });// 第 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 聚合 Futureprivate CompletableFuture<Void> triggerTasks(        CheckpointTriggerRequest request, long timestamp, PendingCheckpoint checkpoint) {finallongcheckpointId= checkpoint.getCheckpointID();final SnapshotType type;if (this.forceFullSnapshot && !request.props.isSavepoint()) {        type = CheckpointType.FULL_CHECKPOINT;    } else {        type = request.props.getCheckpointType();    }// Barrier 传递选项：全量/增量、exactly-once、是否允许非对齐、对齐超时finalCheckpointOptionscheckpointOptions= CheckpointOptions.forConfig(            type,            checkpoint.getCheckpointStorageLocation().getLocationReference(),            isExactlyOnceMode,            unalignedCheckpointsEnabled,            alignedCheckpointTimeout);    List<CompletableFuture<Acknowledge>> acks = newArrayList<>();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、检查 isFullyAcknowledgedpublicbooleanreceiveAcknowledgeMessage(        AcknowledgeCheckpoint message, String taskManagerLocationInfo)throws CheckpointException {    ...synchronized (lock) {finalPendingCheckpointcheckpoint= 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(...);            }returntrue;        }// Checkpoint 已不存在（消息来晚了）→ 丢弃 state        ...    }}

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

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

2.6 如何完成

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

// CheckpointCoordinator.java// 所有 Task 确认后调用。finalizeCheckpoint → 存入 CompletedCheckpointStore → reportCompletedCheckpoint → scheduleTriggerRequest → cleanupAfterCompletedCheckpointprivatevoidcompletePendingCheckpoint(PendingCheckpoint pendingCheckpoint)throws CheckpointException {finallongcheckpointId= pendingCheckpoint.getCheckpointID();final CompletedCheckpoint completedCheckpoint;final CompletedCheckpoint lastSubsumed;finalCheckpointPropertiesprops= 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.javaprivatevoidcleanupAfterCompletedCheckpoint(        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));    }}voidsendAcknowledgeMessages(        List<ExecutionVertex> tasksToCommit,long completedCheckpointId,long completedTimestamp,long lastSubsumedCheckpointId) {// 通知 Taskfor (ExecutionVertex ev : tasksToCommit) {Executionee= ev.getCurrentExecutionAttempt();if (ee != null) {            ee.notifyCheckpointOnComplete(                    completedCheckpointId, completedTimestamp, lastSubsumedCheckpointId);        }    }// 通知 OperatorCoordinatorfor (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 → 执行队列下一个privatevoidonTriggerFailure(@Nullable PendingCheckpoint checkpoint,        CheckpointProperties checkpointProperties,        Throwable throwable) {    throwable = ExceptionUtils.stripCompletionException(throwable);try {// 1. abort OperatorCoordinator 的触发        coordinatorsToCheckpoint.forEach(                OperatorCoordinatorCheckpointContext::abortCurrentTriggering);finalCheckpointExceptioncause= 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.javaprivatevoidabortPendingCheckpoint(        PendingCheckpoint pendingCheckpoint,        CheckpointException exception,@Nullablefinal 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();        }    }}

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

1. 1. Task decline：Task 主动上报无法执行快照（如算子正在关闭、对齐限制超出），发送 DeclineCheckpoint 消息，JobManager 收到后调用 abortPendingCheckpoint
2. 2. 超时：createPendingCheckpoint 时在 timer 上调度了 CheckpointCanceller，超过 checkpointTimeout 未完成就 abort
3. 3. 触发 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 在算子间的传播以及对齐/非对齐策略的源码实现。