Flink源码之Checkpoint执行流程
Checkpoint完整流程如上图所示：
- JobMaster的CheckpointCoordinator向所有SourceTask发送RPC触发一次CheckPoint
- SourceTask向下游广播CheckpointBarrier
- SouceTask完成状态快照后向JobMaster发送快照结果
- 非SouceTask在Barrier对齐后完成状态快照向JobMaster发送快照结果
- JobMaster保存SubTask快照结果
- JobMaster收到所有SubTask快照结果后保存快照信息，想SubTask通知Checkpoint完成
以下对整个流程具体说明。
CheckpointCoordinator
JobMaster将JobGraph转换为ExecutionGraph时，如果开启Checkpoint，会为ExecutionGraph生成一个CheckpointCoordinator
DefaultExecutionGraphBuilder.buildGraph//在此会将JobGraph转换为ExecutionGraphDefaultExecutionGraph::newDefaultExecutionGraph::attachJobGraph //创建ExecutionJobVertexDefaultExecutionTopology.fromExecutionGraph //创建ExecutionTopologyDefaultExecutionGraph::enableCheckpointing //创建CheckpointCoordinatorDefaultExecutionGraph::createCheckpointPlanCalculator//创建DefaultCheckpointPlanCalculatorCheckpointCoordinator::new
CheckpointCoordinator封装了StateBackend和CheckpointStorage
StateBackend负责管理状态：
- HashMapStateBackend //内存
- EmbeddedRocksDBStateBackend //内存+磁盘
CheckpointStorage则是负责存储StateBackend管理的状态：
- JobManagerCheckpointStorage //checkpoint state放入JobManager内存
- FileSystemCheckpointStorage //配置state.checkpoints.dir时
在为StreamTask构造SubtaskCheckpointCoordinatorImpl时会调用：
CheckpointStorage::createCheckpointStorage
创建CheckpointStorageAccess用于执行Checkpoint时解析状态存储位置
- MemoryBackendCheckpointStorageAccess //对应JobManagerCheckpointStorage
- FsCheckpointStorageAccess //对应FileSystemCheckpointStorage
CheckpointCoordinator在执行状态快照时会调用
CheckpointStorageAccess::resolveCheckpointStorageLocation
生成CheckpointStreamFactory用于生成读写状态数据流
- MemCheckpointStreamFactory //对应JobManagerCheckpointStorage
- FsCheckpointStreamFactory //对应FileSystemCheckpointStorage
Checkpoint触发流程
JobMaster状态转换为running后，通过CheckpointCoordinator向SourceTask发送TriggerCheckpoint
JobMaster端触发流程
JobMaster::start //RPCServer启动
JobMaster::onStart
JobMaster::startJobExecution
JobMaster::startJobMasterServices //获取RM地址后与RM建立连接
JobMaster::startScheduling
SchedulerBase::startScheduling
DefaultScheduler::startSchedulingInternal
SchedulerBase::transitionToRunningDefaultExecutionGraph::transitionToRunning //调用ExecutionGraph监听器通知状态变化CheckpointCoordinatorDeActivator::jobStatusChanges//触发checkpointCheckpointCoordinator::startCheckpointSchedulerCheckpointCoordinator::scheduleTriggerWithDelay //定时不断触发CheckpointCheckpointCoordinator::triggerCheckpointCheckpointCoordinator::startTriggeringCheckpointDefaultCheckpointPlanCalculator::calculateCheckpointPlan//Plan中会隔离出SourceTask作为作为Trigger Checkpoint的入口CheckpointCoordinator::createPendingCheckpointCheckpointCoordinator::triggerCheckpointRequestCheckpointCoordinator::triggerTasks Execution::triggerCheckpoint //向每个SourceTask发送TriggerCheckpoint请求Execution::triggerCheckpointHelperTaskManagerGateway::triggerCheckpoint//向TaskExecutor发RPC
StreamTask端执行流程
SourceTask
SourceTask由JobMaster RPC直接触发，执行时先广播CheckpointBarrier，然后对状态执行异步快照
TaskExecutor::triggerCheckpoint
Task::triggerCheckpointBarrier
AbstractInvokable::triggerCheckpointAsync
SourceStreamTask::triggerCheckpointAsync
StreamTask::triggerCheckpointAsync
StreamTask::triggerCheckpointAsyncInMailbox
StreamTask::performCheckpoint
SubtaskCheckpointCoordinatorImpl::checkpointStateOperatorChain.broadcastEvent //广播CheckpointBarrier
CheckpointStorage::createCheckpointStorage//为JobId创建CheckpointStorageAccess
SubtaskCheckpointCoordinatorImpl::takeSnapshotSync
CheckpointStorageWorkerView::resolveCheckpointStorageLocation//CheckpointStorageAccess创建 CheckpointStreamFactoryOperatorChain::snapshotState //对每个OperatorRegularOperatorChain::buildOperatorSnapshotFuturesRegularOperatorChain::checkpointStreamOperatorAbstractStreamOperator::snapshotStateStreamOperatorStateHandler::snapshotState//调用Operator/Keyed Backend的snapshotStateSnapshotContextSynchronousImpl::newAbstractUdfStreamOperator::snapshotState //调用UDF中snapshotState方法，一般用于更新OperatorStateDefaultOperatorStateBackend::snapshotSnapshotStrategyRunner::snapshotDefaultOperatorStateBackendSnapshotStrategy::syncPrepareResources//深copy operator state,便于后续进行异步快照DefaultOperatorStateBackendSnapshotStrategy::asyncSnapshot//异步快照 CheckpointStateOutputStream::closeAndGetHandleOperatorStreamStateHandle::new //包装元信息及数据StreamStateHandleHeapKeyedStateBackend::snapshotSnapshotStrategyRunner::snapshotHeapSnapshotStrategy::syncPrepareResourcesHeapSnapshotStrategy::asyncSnapshot //采用COWSateTable异步快照CheckpointStateOutputStream::closeAndGetHandleKeyGroupsStateHandle::new //包装KeyGroup及数据StreamStateHandle
SubtaskCheckpointCoordinatorImpl::finishAndReportAsync //向JobMaster发送checkpoint的结果AsyncCheckpointRunnable::new AsyncCheckpointRunnable::runAsyncCheckpointRunnable::finalizeNonFinishedSnapshotsOperatorSnapshotFinalizer::new //等待TaskSnapshot状态信息序列化完成AsyncCheckpointRunnable::reportCompletedSnapshotStatesTaskStateManagerImpl::reportTaskStateSnapshotsRpcCheckpointResponder::acknowledgeCheckpoint//向JobMaster发送Ack，带上State信息
非SourceTask
在StreamTask启动后调用StreamTask::processInput不断读取数据进行处理， 非SourceTask在收到上游的CheckpointBarrier对齐后触发Checkpoint,
StreamTask::processInput
StreamOneInputProcessor::processInput
StreamTaskNetworkInput::emitNext(StreamTaskNetworkOutput)
AbstractStreamTaskNetworkInput::emitNext //循环不断从buffer中读取StreamElement
处理CheckpointedInputGate::pollNextCheckpointedInputGate::handleEventSingleCheckpointBarrierHandler::processBarrierSingleCheckpointBarrierHandler::markCheckpointAlignedAndTransformStateWaitingForFirstBarrier::barrierReceivedAbstractAlignedBarrierHandlerState::barrierReceivedSingleCheckpointBarrierHandler.ControllerImpl::allBarriersReceived//判断对齐AbstractAlignedBarrierHandlerState::triggerGlobalCheckpointSingleCheckpointBarrierHandler.ControllerImpl::triggerGlobalCheckpointSingleCheckpointBarrierHandler::triggerCheckpointCheckpointBarrierHandler::notifyCheckpoint //触发StreamTask CheckpointStreamTask::triggerCheckpointOnBarrierStreamTask::performCheckpoint //后续调用过程与SourceTask一样SubtaskCheckpointCoordinatorImpl::checkpointState
根据调用栈看出，非SourceStreamTask执行Checkpoint只是触发时机不同，SourceTask由JobMaster RPC定时不断触发，非SourceTask则是在上游的CheckpointBarrier对齐后触发Checkpoint，最终执行逻辑都是将当前算子的信息写入CheckpointStorage后向JobMaster发送确认信息。
StreamTask向JobMaster ACK信息中包含状态元信息及StreamStateHandle，根据状态存储位置分为：
- ByteStreamStateHandle //对应JobManagerCheckpointStorage,将状态序列化为byte[]发送给JobMaster
- FileStateHandle //对应FileSystemCheckpointStorage,将状态写入文件系统后将文件路径发送给JobMaster
JobMaster端完成流程
JobMaster收到StreamTask的acknowledgeCheckpoint后：
JobMaster::acknowledgeCheckpoint
SchedulerBase::acknowledgeCheckpoint
ExecutionGraphHandler::acknowledgeCheckpoint
CheckpointCoordinator::receiveAcknowledgeMessagePendingCheckpoint::acknowledgeTask //某一个Task的确认PendingCheckpoint::updateOperatorState//更新SubTask状态信息CheckpointCoordinator::completePendingCheckpoint//所有Task Ack后PendingCheckpoint::finalizeCheckpointCheckpoints.storeCheckpointMetadata//保存CheckpointMetadataCompletedCheckpoint::newCheckpointCoordinator::sendAcknowledgeMessages//向Task通知Checkpoint完成消息ExecutionVertex::notifyCheckpointCompleteTaskManagerGateway.notifyCheckpointComplete
JobMaster收到所有StreamTask的Checkpoint状态信息后，标志一次Checkpoint完成，这时会通知StreamTask CheckPoint完成消息，便于SubTask监听Checkpoint完成后做后续动作。