Flink Checkpoint源码浅析

1. JobManager 端checkpoint调度

dispatcher分发任务后会启动相应的jobMaster, 在创建jobMaster 构建过程中会执行jobGraph -> executeGraph的转换，源码如下：

// JobMaster类
public JobMaster(
            RpcService rpcService,
            JobMasterConfiguration jobMasterConfiguration,
            ...)
            throws Exception {
    ...
    this.jobManagerJobMetricGroup = jobMetricGroupFactory.create(jobGraph);
            this.schedulerNG = createScheduler(executionDeploymentTracker, jobManagerJobMetricGroup);
            this.jobStatusListener = null;
    ...
}
// SchedulerBase类
public SchedulerBase(
            final Logger log,
            final JobGraph jobGraph,
            final BackPressureStatsTracker backPressureStatsTracker,
            ...)
            throws Exception {
    ...
        this.executionGraph =
                createAndRestoreExecutionGraph(
                        jobManagerJobMetricGroup,
                        checkNotNull(shuffleMaster),
                        checkNotNull(partitionTracker),
                        checkNotNull(executionDeploymentTracker),
                        initializationTimestamp);
    ...
}
private ExecutionGraph createExecutionGraph(JobManagerJobMetricGroup currentJobManagerJobMetricGroup,
            ShuffleMaster shuffleMaster,
            final JobMasterPartitionTracker partitionTracker,
            ExecutionDeploymentTracker executionDeploymentTracker,
            long initializationTimestamp)
            throws JobExecutionException, JobException {
    ...
        return ExecutionGraphBuilder.buildGraph(null,
                jobGraph,
                jobMasterConfiguration,
                ...);
    ...
    
}

createAndRestoreExecutionGraph()方法调用了createExecutionGraph()方法最终使用ExecutionGraphBuilder进行了ExecuteGraph的生成。

在构建ExecutionGraph过程中（ExecutionGraphBuilder.buildGraph()方法），会调用ExecutionGraph.enableCheckpointing()方法，这个方法不管任务里有没有设置checkpoint都会调用的。在enableCheckpointing()方法里会创建CheckpointCoordinator，这是负责checkpoint的核心实现类，同时会给job添加一个监听器CheckpointCoordinatorDeActivator（只有设置了checkpoint才会注册这个监听器），CheckpointCoordinatorDeActivator负责checkpoint的启动和停止。源码如下：

// ExecutionGraphBuilder类
public static ExecutionGraph buildGraph(
            @Nullable ExecutionGraph prior,
            JobGraph jobGraph,
            Configuration jobManagerConfig,
            ...)
            throws JobExecutionException, JobException {
    ...
        // configure the state checkpointing
        JobCheckpointingSettings snapshotSettings = jobGraph.getCheckpointingSettings();
        if (snapshotSettings != null) {
            List triggerVertices =
                    idToVertex(snapshotSettings.getVerticesToTrigger(), executionGraph);
​
            List ackVertices =
                    idToVertex(snapshotSettings.getVerticesToAcknowledge(), executionGraph);
​
            List confirmVertices =
                    idToVertex(snapshotSettings.getVerticesToConfirm(), executionGraph);
            // 一系列的checkpoint设置，包括statebackend, user-define hook, checkpointIdCounter等
            ...
            executionGraph.enableCheckpointing(
                    chkConfig,
                    triggerVertices,
                    ackVertices,
                    confirmVertices,
                    hooks,
                    checkpointIdCounter,
                    completedCheckpoints,
                    rootBackend,
                    checkpointStatsTracker);
    ...    
}

在 build graph 时确定了 triggerVertices ( 用来触发 chekcpoint)，ackVertices ( 用来接收 checkpoint 已经完成的报告 )以及 confirmVertices ( 用来确认 checkpoint 已经完成 )。

executionGraph.enableCheckpointing(）中做了一些checkpoint相关类的初始化操作，以及checkpoint状态监听器的注册。在JobManager端开始进行任务调度的时候，会对job的状态进行转换，由CREATED转成RUNNING，实现在transitionState()方法中，在这个过程中刚才设置的job监听器CheckpointCoordinatorDeActivator就开始启动checkpoint的定时任务了，调用链为ExecutionGraph.transitionToRunning() -> transitionState() -> notifyJobStatusChange() -> CheckpointCoordinatorDeActivator.jobStatusChanges() -> CheckpointCoordinator.startCheckpointScheduler()源码如下：

public void transitionToRunning() {
    if (!transitionState(JobStatus.CREATED, JobStatus.RUNNING)) {
        throw new IllegalStateException(
                "Job may only be scheduled from state " + JobStatus.CREATED);
    }
}
​
private boolean transitionState(JobStatus current, JobStatus newState, Throwable error) {
    ...
    if (state == current) {
        notifyJobStatusChange(newState, error);
        return true;
    }
    ...
}
​
private void notifyJobStatusChange(JobStatus newState, Throwable error) {
    if (jobStatusListeners.size() > 0) {
        final long timestamp = System.currentTimeMillis();
        final Throwable serializedError = error == null ? null : new SerializedThrowable(error);
​
        for (JobStatusListener listener : jobStatusListeners) {
            try {
                listener.jobStatusChanges(getJobID(), newState, timestamp, serializedError);
            } catch (Throwable t) {
                LOG.warn("Error while notifying JobStatusListener", t);
            }
        }
    }
}
​
// CheckpointCoordinatorDeActivator类
@Override
    public void jobStatusChanges(
            JobID jobId, JobStatus newJobStatus, long timestamp, Throwable error) {
        if (newJobStatus == JobStatus.RUNNING) {
            // start the checkpoint scheduler
            coordinator.startCheckpointScheduler();
        } else {
            // anything else should stop the trigger for now
            coordinator.stopCheckpointScheduler();
        }
    }

CheckpointCoordinator会部署一个定时任务，用于周期性的触发checkpoint，这个定时任务就是ScheduledTrigger类。

public void startCheckpointScheduler() {
    synchronized (lock) {
        if (shutdown) {
            throw new IllegalArgumentException("Checkpoint coordinator is shut down");
        }
​
        // make sure all prior timers are cancelled
        stopCheckpointScheduler();
​
        periodicScheduling = true;
        currentPeriodicTrigger = scheduleTriggerWithDelay(getRandomInitDelay());
    }
}
​
private ScheduledFuture scheduleTriggerWithDelay(long initDelay) {
    return timer.scheduleAtFixedRate(
        new ScheduledTrigger(), initDelay, baseInterval, TimeUnit.MILLISECONDS);
}
​
public CompletableFuture triggerCheckpoint(
            CheckpointProperties props,
            @Nullable String externalSavepointLocation,
            boolean isPeriodic) {
​
    if (props.getCheckpointType().getPostCheckpointAction() == PostCheckpointAction.TERMINATE
        && !(props.isSynchronous() && props.isSavepoint())) {
        return FutureUtils.completedExceptionally(
            new IllegalArgumentException(
                "Only synchronous savepoints are allowed to advance the watermark to MAX."));
    }
​
    CheckpointTriggerRequest request =
        new CheckpointTriggerRequest(props, externalSavepointLocation, isPeriodic);
    // 首先做一些前置校验，看是否能触发checkpoint，主要就是检查最大并发checkpoint数，checkpoint间隔时间
    // 在积压（如果有）的checkpoint中选一个进行处理
    chooseRequestToExecute(request).ifPresent(this::startTriggeringCheckpoint);
    return request.onCompletionPromise;
}
​
private void startTriggeringCheckpoint(CheckpointTriggerRequest request) {
    try {
        synchronized (lock) {
            preCheckGlobalState(request.isPeriodic);
        }
        // 找出需要发送checkpoint消息的task(即tasksToTrigger，由生成JobGraph时生成，由所有不包含输入的顶点组成)放入executions
        // Check if all tasks that we need to trigger are running. If not, abort the checkpoint.
        final Execution[] executions = getTriggerExecutions();
        // 找出需要返回checkpoint的ack反馈信息的task放入ackTasks，并将其作为构造PendingCheckpoint的参数
        // Check if all tasks that need to acknowledge the checkpoint are running. If not, abort the checkpoint
        final Map ackTasks = getAckTasks();
        
        // 创建PendingCheckpoint, 用户自定义hook触发...
        ... 
        // no exception, no discarding, everything is OK
            final long checkpointId =
            checkpoint.getCheckpointId();
            snapshotTaskState(
                timestamp,
                checkpointId,
                checkpoint.getCheckpointStorageLocation(),
                request.props,
                executions);
        
            coordinatorsToCheckpoint.forEach(
                (ctx) ->
                ctx.afterSourceBarrierInjection(
                    checkpointId));
        ...
    } catch (Throwable throwable) {
        onTriggerFailure(request, throwable);
    }
}
​
private void snapshotTaskState(
            long timestamp,
            long checkpointID,
            CheckpointStorageLocation checkpointStorageLocation,
            CheckpointProperties props,
            Execution[] executions) {
    ...
    // send the messages to the tasks that trigger their checkpoint
    for (Execution execution : executions) {
        // 两者底层调用的是同一个方法，只有语义上的区别
        if (props.isSynchronous()) {
            execution.triggerSynchronousSavepoint(checkpointID, timestamp, checkpointOptions);
        } else {
            execution.triggerCheckpoint(checkpointID, timestamp, checkpointOptions);
        }
    }
}

Execution.triggerCheckpoint()就是远程调用TaskManager的triggerCheckpoint()方法:

private void triggerCheckpointHelper(
        long checkpointId, long timestamp, CheckpointOptions checkpointOptions) {
    ...
    final LogicalSlot slot = assignedResource;
​
    if (slot != null) {
        final TaskManagerGateway taskManagerGateway = slot.getTaskManagerGateway();
​
        taskManagerGateway.triggerCheckpoint(
                attemptId, getVertex().getJobId(), checkpointId, timestamp, checkpointOptions);
    } else {
        LOG.debug(
                "The execution has no slot assigned. This indicates that the execution is no longer running.");
    }
}

2. SourceStreamTask的Checkpoint执行

TaskManager的triggerCheckpoint()方法首先获取到source task（即SourceStreamTask），调用Task.triggerCheckpointBarrier()，triggerCheckpointBarrier()会异步的去执行一个独立线程，这个线程来负责source task的checkpoint执行。

// TaskExecutor类
public CompletableFuture triggerCheckpoint(
        ExecutionAttemptID executionAttemptID,
        long checkpointId,
        long checkpointTimestamp,
        CheckpointOptions checkpointOptions) {
    log.debug(
            "Trigger checkpoint {}@{} for {}.",
            checkpointId,
            checkpointTimestamp,
            executionAttemptID);
    ...
    final Task task = taskSlotTable.getTask(executionAttemptID);
​
    if (task != null) {
        task.triggerCheckpointBarrier(checkpointId, checkpointTimestamp, checkpointOptions);
​
        return CompletableFuture.completedFuture(Acknowledge.get());
    } else {
        ...
    }
}
​
// Task类
public void triggerCheckpointBarrier(
    final long checkpointID,
    final long checkpointTimestamp,
    final CheckpointOptions checkpointOptions) {
    ...
    try {
        // invokable 事实上就是 StreamTask 类，而 StreamTask 也将委托给更具体的类，直到业务代码
        invokable.triggerCheckpointAsync(checkpointMetaData, checkpointOptions);
    }
    ...
}

由 invokable 调用 triggerCheckpoint。由于 trigger task 都是 source operator chain 所以进入 SourceStreamTask:

public Future triggerCheckpointAsync(
        CheckpointMetaData checkpointMetaData, CheckpointOptions checkpointOptions) {
    if (!externallyInducedCheckpoints) {
        return super.triggerCheckpointAsync(checkpointMetaData, checkpointOptions);
    } else {
        // we do not trigger checkpoints here, we simply state whether we can trigger them
        synchronized (lock) {
            return CompletableFuture.completedFuture(isRunning());
        }
    }
}

checkpoint的核心实现在StreamTask.performCheckpoint()方法中，该方法主要有三个步骤

1、在checkpoint之前做一些准备工作，通常情况下operator在这个阶段是不做什么操作的

2、立即向下游广播CheckpointBarrier，以便使下游的task能够及时的接收到CheckpointBarrier也开始进行checkpoint的操作

3、开始进行状态的快照，即checkpoint操作。

在进行performCheckpoint()时，task任务线程是不能够进行数据处理的， checkpoint和任务处理使用的是同一把锁:

public Future triggerCheckpointAsync(CheckpointMetaData checkpointMetaData, CheckpointOptions checkpointOptions) {
    ...
    triggerCheckpointAsyncInMailbox(checkpointMetaData, checkpointOptions));
    ...
}

private boolean triggerCheckpointAsyncInMailbox(
            CheckpointMetaData checkpointMetaData, CheckpointOptions checkpointOptions)
    throws Exception {
    ...
    subtaskCheckpointCoordinator.initCheckpoint(
                    checkpointMetaData.getCheckpointId(), checkpointOptions);

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

// SubtaskCheckpointCoordinatorImpl类
public void checkpointState(
            CheckpointMetaData metadata,
            CheckpointOptions options,
            CheckpointMetricsBuilder metrics,
            OperatorChain operatorChain,
            Supplier isRunning)
            throws Exception {
    
    //校验checkpoint是否需要终止
    // Step (0): Record the last triggered checkpointId and abort the sync phase of checkpoint
    // if necessary.
    lastCheckpointId = metadata.getCheckpointId();
    if (checkAndClearAbortedStatus(metadata.getCheckpointId())) {
        // broadcast cancel checkpoint marker to avoid downstream back-pressure due to
        // checkpoint barrier align.
        operatorChain.broadcastEvent(new CancelCheckpointMarker(metadata.getCheckpointId()));
        LOG.info(
            "Checkpoint {} has been notified as aborted, would not trigger any checkpoint.",
            metadata.getCheckpointId());
        return;
    }
    
    
    // Step (1): Prepare the checkpoint, allow operators to do some pre-barrier work.
    //           The pre-barrier work should be nothing or minimal in the common case.
    // 一般无逻辑
    operatorChain.prepareSnapshotPreBarrier(metadata.getCheckpointId());
    
    // Step (2): Send the checkpoint barrier downstream
    // 封装优先级buffer后add到ResultSubpartition的PrioritizedDeque队列中，更新buffer和backlog数
    // 当notifyDataAvailable=true时 通知下游消费
    // 下游CheckpointedInputGate拿到buffer后匹配到是checkpoint事件做出相应动作
    operatorChain.broadcastEvent(
        new CheckpointBarrier(metadata.getCheckpointId(), metadata.getTimestamp(), options),
        options.isUnalignedCheckpoint());
    
    // 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());
    }

    // Step (4): Take the state snapshot. This should be largely asynchronous, to not impact
    // progress of the
    // streaming topology

    Map snapshotFutures =
        new HashMap<>(operatorChain.getNumberOfOperators());
    try {
        // takeSnapshotSync 执行checkpoint核心逻辑的入口
        if (takeSnapshotSync(
            snapshotFutures, metadata, metrics, options, operatorChain, isRunning)) {
            // finishAndReportAsync 完成snapshot后，向jobMaster发送报告
            finishAndReportAsync(snapshotFutures, metadata, metrics, isRunning);
        } else {
            cleanup(snapshotFutures, metadata, metrics, new Exception("Checkpoint declined"));
        }
    } catch (Exception ex) {
        cleanup(snapshotFutures, metadata, metrics, ex);
        throw ex;
    }
}

private boolean takeSnapshotSync(
            Map operatorSnapshotsInProgress,
            CheckpointMetaData checkpointMetaData,
            CheckpointMetricsBuilder checkpointMetrics,
            CheckpointOptions checkpointOptions,
            OperatorChain operatorChain,
            Supplier isRunning)
            throws Exception {
    ...
    // 存储checkpoint的位置(Memory/FS/RockDB)
    CheckpointStreamFactory storage =
        checkpointStorage.resolveCheckpointStorageLocation(
        checkpointId, checkpointOptions.getTargetLocation());

    try {
        for (StreamOperatorWrapper operatorWrapper :
             operatorChain.getAllOperators(true)) {
            if (!operatorWrapper.isClosed()) {
                operatorSnapshotsInProgress.put(
                    operatorWrapper.getStreamOperator().getOperatorID(),
                    // 执行checkpoint入口
                    buildOperatorSnapshotFutures(
                        checkpointMetaData,
                        checkpointOptions,
                        operatorChain,
                        operatorWrapper.getStreamOperator(),
                        isRunning,
                        channelStateWriteResult,
                        storage));
            }
        }
    } finally {
        checkpointStorage.clearCacheFor(checkpointId);
    }
    ...
}

//StreamOperatorStateHandler类
void snapshotState(
    CheckpointedStreamOperator streamOperator,
    Optional> timeServiceManager,
    String operatorName,
    long checkpointId,
    long timestamp,
    CheckpointOptions checkpointOptions,
    CheckpointStreamFactory factory,
    OperatorSnapshotFutures snapshotInProgress,
    StateSnapshotContextSynchronousImpl snapshotContext,
    boolean isUsingCustomRawKeyedState)
    throws CheckpointException {
    try {
        ...
        //执行需要持久化state的操作
        //比如map操作，它生成的是StreamMap属于AbstractUdfStreamOperator子类，里面封装了snapshotState逻辑，如果没实现ck接口就跳过此步骤
        streamOperator.snapshotState(snapshotContext);

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

        if (null != operatorStateBackend) {
            snapshotInProgress.setOperatorStateManagedFuture(
                operatorStateBackend.snapshot(
                    checkpointId, timestamp, factory, checkpointOptions));
        }

        if (null != keyedStateBackend) {
            snapshotInProgress.setKeyedStateManagedFuture(
                keyedStateBackend.snapshot(
                    checkpointId, timestamp, factory, checkpointOptions));
        }
    } 
    ...
}

如果用户实现了Checkpoint接口则会持久化到指定的stateBackend中反之略过...

这里以AbstractUdfStreamOperator为例（map，filter等Operator都继承了该abstract类）:

// AbstractUdfStreamOperator类
public void snapshotState(StateSnapshotContext context) throws Exception {
    super.snapshotState(context);
    //判断userFunction是否属于CheckpointedFunction或者ListCheckpointed的实例
    //如果是则调用用户实现的snapshotState执行相关逻辑
    //比如FlinkKafkaConsumerBase则自己实现了CheckpointedFunction的接口
    StreamingFunctionUtils.snapshotFunctionState(
            context, getOperatorStateBackend(), userFunction);
}

// StreamingFunctionUtils类
public static void snapshotFunctionState(
    StateSnapshotContext context, OperatorStateBackend backend, Function userFunction)
    throws Exception {

    Preconditions.checkNotNull(context);
    Preconditions.checkNotNull(backend);

    while (true) {
        // 校验用户是否有自定义checkpoint逻辑并执行用户自定义逻辑
        if (trySnapshotFunctionState(context, backend, userFunction)) {
            break;
        }

        // inspect if the user function is wrapped, then unwrap and try again if we can snapshot
        // the inner function
        if (userFunction instanceof WrappingFunction) {
            userFunction = ((WrappingFunction) userFunction).getWrappedFunction();
        } else {
            break;
        }
    }
}

private static boolean trySnapshotFunctionState(
    StateSnapshotContext context, OperatorStateBackend backend, Function userFunction)
    throws Exception {
	// 判断用户是否实现CheckpointedFunction接口
    if (userFunction instanceof CheckpointedFunction) {
        // 执行用户自定义的snapshot逻辑
        ((CheckpointedFunction) userFunction).snapshotState(context);

        return true;
    }

    if (userFunction instanceof ListCheckpointed) {
        // ListCheckpointed已废弃不再多说
        ...
    }
    return false;
}

3. Task上报checkpoint信息

整个快照生成完毕，最后Flink会调用finishAndReportAsync向Master发送完成报告：

private void finishAndReportAsync(
        Map snapshotFutures,
        CheckpointMetaData metadata,
        CheckpointMetricsBuilder metrics,
        Supplier isRunning) {
    // we are transferring ownership over snapshotInProgressList for cleanup to the thread,
    // active on submit
    asyncOperationsThreadPool.execute(
            new AsyncCheckpointRunnable(
                    snapshotFutures,
                    metadata,
                    metrics,
                    System.nanoTime(),
                    taskName,
                    registerConsumer(),
                    unregisterConsumer(),
                    env,
                    asyncExceptionHandler,
                    isRunning));
}

最终会调用到CheckpointCoordinator.receiveAcknowledgeMessage()方法:

public boolean receiveAcknowledgeMessage(AcknowledgeCheckpoint message, String taskManagerLocationInfo)
            throws CheckpointException {
    ...
    synchronized (lock) {
        // we need to check inside the lock for being shutdown as well, otherwise we
        // get races and invalid error log messages
        if (shutdown) {
            return false;
        }

        final PendingCheckpoint checkpoint = pendingCheckpoints.get(checkpointId);

        if (checkpoint != null && !checkpoint.isDisposed()) {

            switch (checkpoint.acknowledgeTask(
                message.getTaskExecutionId(),
                message.getSubtaskState(),
                message.getCheckpointMetrics())) {
                case SUCCESS:
                    LOG.debug(
                        "Received acknowledge message for checkpoint {} from task {} of job {} at {}.",
                        checkpointId,
                        message.getTaskExecutionId(),
                        message.getJob(),
                        taskManagerLocationInfo);

                    if (checkpoint.isFullyAcknowledged()) {
                        completePendingCheckpoint(checkpoint);
                    }
                    break;
                    ...
            }
            ...
        }
        ...
    }
}

JobManager完成所有task的ack之后，会做以下操作：

1、将PendingCheckpoint 转成CompletedCheckpoint，标志着checkpoint过程完成，CompletedCheckpoint里包含了checkpoint的元数据信息，包括checkpoint的路径地址，状态数据大小等等，同时也会将元数据信息进行持久化，也会把过期的checkpoint数据给删除

2、通知所有的task进行commit操作。

private void completePendingCheckpoint(PendingCheckpoint pendingCheckpoint){
    ...
        // 生成CompeletePoint
        completedCheckpoint =
        pendingCheckpoint.finalizeCheckpoint(
        checkpointsCleaner, this::scheduleTriggerRequest, executor);
    	// 持久化checkpoint到state backend
    	completedCheckpointStore.addCheckpoint(
        	completedCheckpoint, checkpointsCleaner, this::scheduleTriggerRequest);
    ...
        // 通知taskcheckpoint完成
        // send the "notify complete" call to all vertices, coordinators, etc.
        sendAcknowledgeMessages(checkpointId, completedCheckpoint.getTimestamp());
}

private void sendAcknowledgeMessages(long checkpointId, long timestamp) {
    // commit tasks
    for (ExecutionVertex ev : tasksToCommitTo) {
        Execution ee = ev.getCurrentExecutionAttempt();
        if (ee != null) {
            ee.notifyCheckpointComplete(checkpointId, timestamp);
        }
    }

    // commit coordinators
    for (OperatorCoordinatorCheckpointContext coordinatorContext : coordinatorsToCheckpoint) {
        coordinatorContext.notifyCheckpointComplete(checkpointId);
    }
}

4. JobManager通知Task进行commit

task在接收到消息之后会调用Task.notifyCheckpointComplete()方法，最后会调用StreamOperator.notifyCheckpointComplete()，一般来说不做什么操作。但是像AbstractUdfStreamOperator这种的可能还会由一些其他操作：

// TaskExecutor类
public CompletableFuture confirmCheckpoint(
        ExecutionAttemptID executionAttemptID, long checkpointId, long checkpointTimestamp) {
    ...
    final Task task = taskSlotTable.getTask(executionAttemptID);

    if (task != null) {
        task.notifyCheckpointComplete(checkpointId);

        return CompletableFuture.completedFuture(Acknowledge.get());
    } 
    ...
}

// Task类
public void notifyCheckpointComplete(final long checkpointID) {
    // invokable 事实上就是 StreamTask 类，而 StreamTask 也将委托给更具体的类，直到业务代码
    final AbstractInvokable invokable = this.invokable;

        if (executionState == ExecutionState.RUNNING && invokable != null) {
            try {
                invokable.notifyCheckpointCompleteAsync(checkpointID);
            }...
        }
}

AbstractUdfStreamOperator主要是针对用户自定义函数的operator，像StreamMap，StreamSource等等，如果用户定义的Function实现了CheckpointListener接口，则会进行额外的一些处理，例如FlinkKafkaConsumerBase会向kafka提交消费的offset，TwoPhaseCommitSinkFunction类会进行事务的提交，例如FlinkKafkaProducer。

5. 非SourceStreamTask的checkpoint实现

上述是source task的checkpoint实现，source task的checkpoint是由JobManager来触发的，source task会向下游广播发送CheckpointBarrier，那么下游的task就会接收到source task发送的CheckpointBarrier，checkpoint的起始位置也在接收到CheckpointBarrier。非SourceTask一直通过循环从上游读取消息，当接收一条消息后，会对消息类型进行判断，如果是CheckpointBarrier类型的消息则会进一步判断是需要对齐或是进行checkpoint。该逻辑在 CheckpointInputGate#pollNext()方法中进行：

public Optional pollNext() throws IOException, InterruptedException {
    Optional next = inputGate.pollNext();

    if (!next.isPresent()) {
        return handleEmptyBuffer();
    }

    BufferOrEvent bufferOrEvent = next.get();

    if (bufferOrEvent.isEvent()) {
        return handleEvent(bufferOrEvent);
    } else if (bufferOrEvent.isBuffer()) {
        barrierHandler.addProcessedBytes(bufferOrEvent.getBuffer().getSize());
    }
    return next;
}

private Optional handleEvent(BufferOrEvent bufferOrEvent)
            throws IOException, InterruptedException {
    Class eventClass = bufferOrEvent.getEvent().getClass();
    if (eventClass == CheckpointBarrier.class) {
        CheckpointBarrier checkpointBarrier = (CheckpointBarrier) bufferOrEvent.getEvent();
        // 处理barrier
        barrierHandler.processBarrier(checkpointBarrier, bufferOrEvent.getChannelInfo());
    }
    ...
}

// SingleCheckpointBarrierHandler类 barrier对齐（CheckpointBarrierTracker类是 at last once）
public void processBarrier(CheckpointBarrier barrier, InputChannelInfo channelInfo)
            throws IOException {
    long barrierId = barrier.getId();
    LOG.debug("{}: Received barrier from channel {} @ {}.", taskName, channelInfo, barrierId);

	// barrier滞后，已经在处理新checkpoint或者checkpoint已经超时置为非pending状态，丢弃原有的，释放通道消费
    if (currentCheckpointId > barrierId
        || (currentCheckpointId == barrierId && !isCheckpointPending())) {
        controller.obsoleteBarrierReceived(channelInfo, barrier);
        return;
    }

    // 当前checkpoint滞后，收到新checkpoint的barrier，开始新的checkpoint
    if (currentCheckpointId < barrierId) {
        if (isCheckpointPending()) {
            // cancel 旧的checkpoint
            cancelSubsumedCheckpoint(barrierId);
        }
		
        if (getNumOpenChannels() == 1) {
            // 如果上游通道数只有一个，直接触发checkpoint
            markAlignmentStartAndEnd(barrierId, barrier.getTimestamp());
        } else {
            // 上游有多个通道，开始对齐
            markAlignmentStart(barrierId, barrier.getTimestamp());
        }
        currentCheckpointId = barrierId;
        numBarriersReceived = 0;
        allBarriersReceivedFuture = new CompletableFuture<>();
        try {
            // 首次收到barrier处理
            if (controller.preProcessFirstBarrier(channelInfo, barrier)) {
                LOG.debug(
                    "{}: Triggering checkpoint {} on the first barrier at {}.",
                    taskName,
                    barrier.getId(),
                    barrier.getTimestamp());
                notifyCheckpoint(barrier);
            }
        } catch (CheckpointException e) {
            abortInternal(barrier.getId(), e);
            return;
        }
    }
	// 接收barrier并阻塞相应的channel
    controller.barrierReceived(channelInfo, barrier);

    if (currentCheckpointId == barrierId) {
        if (++numBarriersReceived == numOpenChannels) {
            if (getNumOpenChannels() > 1) {
                markAlignmentEnd();
            }
            numBarriersReceived = 0;
            // 所有barrier均已到达，处理最后一个barrier
            if (controller.postProcessLastBarrier(channelInfo, barrier)) {
                LOG.debug(
                    "{}: Triggering checkpoint {} on the last barrier at {}.",
                    taskName,
                    barrier.getId(),
                    barrier.getTimestamp());
                // 开始checkpoint，实际还是调用之前的StreamTask.performCheckpoint()方法，后续跟以上source checkpoint一致
                notifyCheckpoint(barrier);
            }
            allBarriersReceivedFuture.complete(null);
        }
    }
}

总结： 总的来说checkpoint是通过job状态的变更来启动，接下来找到source task 进行ck，同时将barrier发送到下游算子通知他们开始自己的ck，算子完成后进行回调通知。过称还算比较清晰，细节没有细抠。以上仅仅对checkpoint的barrier过程做了一次简单的分析，具体的状态持久化过程没有涉及，也没有对非对齐barrier做解析，有兴趣的可以自己看看，后续可能会有相应的文章

参考文章：

Flink源码分析——Checkpoint源码分析(一) - 知乎

Flink1.12源码解读——Checkpoint详细执行过程_按时吃早饭ABC的博客-CSDN博客