Flink Checkpoint 流程

Checkpoint 源碼流程：

　　Flink MiniCluster 啟動流程 

　　Flink CheckpointCoordinator 啟動流程  

　　Flink Checkpoint 流程 

先貼段簡單的代碼

val kafkaSource = new FlinkKafkaConsumer[String]("kafka_offset", new SimpleStringSchema(), prop)
val kafkaSource1 = new FlinkKafkaConsumer[String]("kafka_offset", new SimpleStringSchema(), prop)
val kafkaProducer = new FlinkKafkaProducer[String]("kafka_offset_out", new SimpleStringSchema(), prop)
val source = env
  .addSource(kafkaSource)
  .setParallelism(1)
  .name("source")
val source1 = env
  .addSource(kafkaSource1)
  .setParallelism(1)
  .name("source1")

source.union(source1) .map(node => {
    node.toString + ",flinkx"
  })
  .addSink(kafkaProducer)

很簡單，就是讀Kafka，再寫回kafka，主要是Checkpoint 的流程，代碼在這里就不重要了

-------------------------------------------

一個簡化的 Checkpoint 流圖

 

1、CheckpointCoordicator tirgger checkpoint 到 source
2、Source
　　1、生成並廣播 CheckpointBarrier
　　2、Snapshot state（完成后 ack Checkpoint 到 CheckpointCoordicator）
3、Map
　　1、接收到 CheckpointBarrier
　　2、廣播 CheckpointBarrier
　　3、Snapshot state（完成后 ack Checkpoint 到 CheckpointCoordicator）
4、Sink
　　1、接收到 CheckpointBarrier
　　2、Snapshot state（完成后 ack Checkpoint 到 CheckpointCoordicator）
5、CheckpointCoordicator 接收到 所有 ack
　　1、給所有算子發 notifyCheckpointComplete
6、Source、Map、Sink 收到 notifyCheckpointComplete

 

這里有個更好的圖： https://www.cnblogs.com/bethunebtj/p/9168274.html#5-%E4%B8%BA%E6%89%A7%E8%A1%8C%E4%BF%9D%E9%A9%BE%E6%8A%A4%E8%88%AAfault-tolerant%E4%B8%8E%E4%BF%9D%E8%AF%81exactly-once%E8%AF%AD%E4%B9%89

從流程圖上可以看出，Checkpoint 由 CheckpointCoordinator 發起、確認，通過Rpc 通知 Taskmanager 的具體算子完成 Checkpoint 操作。

Checkpoint Timer 啟動

Flink Checkpoint 是由 CheckpointCoordinator 協調啟動，有個內部類來做這個事情

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);
        }
    }
}

JobMaster.java 啟動 Checkpoint 的 timer

private void startScheduling() {
        checkState(jobStatusListener == null);
        // register self as job status change listener
        jobStatusListener = new JobManagerJobStatusListener();
        schedulerNG.registerJobStatusListener(jobStatusListener);
        // 這個會調用到 CheckpointCoordinator.scheduleTriggerWithDelay 方法啟動第一次 Checkpoint，后續就由 CheckpointCoordinator 自己啟動
        schedulerNG.startScheduling();
    }

ScheduledTrigger 由 Timer 定時調用

private ScheduledFuture<?> scheduleTriggerWithDelay(long initDelay) {
    return timer.scheduleAtFixedRate(
        new ScheduledTrigger(),
        initDelay, baseInterval, TimeUnit.MILLISECONDS);
}

/**
 * Executes the given command periodically. The first execution is started after the
 * {@code initialDelay}, the second execution is started after {@code initialDelay + period},
 * the third after {@code initialDelay + 2*period} and so on.
 * The task is executed until either an execution fails, or the returned {@link ScheduledFuture}
 * is cancelled.
 *
 * @param command the task to be executed periodically
 * @param initialDelay the time from now until the first execution is triggered  第一次啟動 Checkpoint 時間
 * @param period the time after which the next execution is triggered            后續的時間間隔
 * @param unit the time unit of the delay and period parameter
 * @return a ScheduledFuture representing the periodic task. This future never completes
 * unless an execution of the given task fails or if the future is cancelled
 */
ScheduledFuture<?> scheduleAtFixedRate(
    Runnable command,
    long initialDelay,
    long period,
    TimeUnit unit);

ScheduledTrigger 開始 checkpoint

ScheduledTrigger.run 方法  調用 triggerCheckpoint 開始執行 checkpoint

public CompletableFuture<CompletedCheckpoint> triggerCheckpoint(boolean isPeriodic) {
    return triggerCheckpoint(checkpointProperties, null, isPeriodic, false);
}

@VisibleForTesting
public CompletableFuture<CompletedCheckpoint> triggerCheckpoint(
        CheckpointProperties props,
        @Nullable String externalSavepointLocation,
        boolean isPeriodic,
        boolean advanceToEndOfTime) {


    CheckpointTriggerRequest request = new CheckpointTriggerRequest(props, externalSavepointLocation, isPeriodic, advanceToEndOfTime);
    requestDecider
        .chooseRequestToExecute(request, isTriggering, lastCheckpointCompletionRelativeTime)
        //  調用 startTriggeringCheckpoint 方法
        .ifPresent(this::startTriggeringCheckpoint);
    return request.onCompletionPromise;
}

private void startTriggeringCheckpoint(CheckpointTriggerRequest request) {

    // 獲取 需要觸發 Checkpoint 的算子
    final Execution[] executions = getTriggerExecutions();
....
    // no exception, no discarding, everything is OK
    final long checkpointId = checkpoint.getCheckpointId();
    snapshotTaskState(
        timestamp,
        checkpointId,
        checkpoint.getCheckpointStorageLocation(),
        request.props,
        executions,
        request.advanceToEndOfTime);

    coordinatorsToCheckpoint.forEach((ctx) -> ctx.afterSourceBarrierInjection(checkpointId));

    onTriggerSuccess();
............
}


private void snapshotTaskState(
        long timestamp,
        long checkpointID,
        CheckpointStorageLocation checkpointStorageLocation,
        CheckpointProperties props,
        Execution[] executions,
        boolean advanceToEndOfTime) {

        // send the messages to the tasks that trigger their checkpoint
        // 給每個 Execution (在這里可以理解為每個 Source 算子，因為Checkpoint 是從 Source 開始的) 發送 trigger checkpoint 消息
        for (Execution execution: executions) {
            if (props.isSynchronous()) {
                execution.triggerSynchronousSavepoint(checkpointID, timestamp, checkpointOptions, advanceToEndOfTime);
            } else {
                execution.triggerCheckpoint(checkpointID, timestamp, checkpointOptions);
            }
        }
    }

Execution.java 的 triggerCheckpoint 方法 調用 triggerCheckpointHelper 方法, 通過 TaskManagerGateway 發送 triggerCheckpoint 的 RPC 請求，

調用 RpcTaskManagerGateway.triggerCheckpoint 方法，然后調用 TaskExecutorGateway 的 triggerCheckpoint 方法（TaskExecutor繼承自 TaskExecutorGateway，就到 TaskManager 端了） 

private void triggerCheckpointHelper(long checkpointId, long timestamp, CheckpointOptions checkpointOptions, boolean advanceToEndOfEventTime) {

    final CheckpointType checkpointType = checkpointOptions.getCheckpointType();
    if (advanceToEndOfEventTime && !(checkpointType.isSynchronous() && checkpointType.isSavepoint())) {
        throw new IllegalArgumentException("Only synchronous savepoints are allowed to advance the watermark to MAX.");
    }

    final LogicalSlot slot = assignedResource;

    if (slot != null) {
        final TaskManagerGateway taskManagerGateway = slot.getTaskManagerGateway();

        taskManagerGateway.triggerCheckpoint(attemptId, getVertex().getJobId(), checkpointId, timestamp, checkpointOptions, advanceToEndOfEventTime);
    } else {
        LOG.debug("The execution has no slot assigned. This indicates that the execution is no longer running.");
    }
}

到這里， JobManager 端觸發 checkpoint 就完成了，下面就是 TaskManager 端接收 triggerCheckpoint 消息了

TaskManager 接收 triggerCheckpoint 消息

從上面可以知道，JobManager 給 TaskManager 發送 Rpc 請求，調用 RpcTaskManagerGateway.triggerCheckpoint 發送checkpoint 的 Rpc 到 TaskManager，TaskManager 接收到 Rpc 后，會反射到 TaskExecutor 的 triggerCheckpoint 方法，這里就進入 TaskManager 里面了

@Override
public CompletableFuture<Acknowledge> triggerCheckpoint(
        ExecutionAttemptID executionAttemptID,
        long checkpointId,
        long checkpointTimestamp,
        CheckpointOptions checkpointOptions,
        boolean advanceToEndOfEventTime) {
    log.debug("Trigger checkpoint {}@{} for {}.", checkpointId, checkpointTimestamp, executionAttemptID);

    final CheckpointType checkpointType = checkpointOptions.getCheckpointType();
    if (advanceToEndOfEventTime && !(checkpointType.isSynchronous() && checkpointType.isSavepoint())) {
        throw new IllegalArgumentException("Only synchronous savepoints are allowed to advance the watermark to MAX.");
    }
    // 使用 operator id 獲取 job 對應的 Source 算子 （如果有多個 Source，JobManager 端會發送兩個 Rpc 請求，TaskManager 也是執行兩次）
    final Task task = taskSlotTable.getTask(executionAttemptID);

    if (task != null) {
        // 對 task 觸發 一個 Checkpoint Barrier
        task.triggerCheckpointBarrier(checkpointId, checkpointTimestamp, checkpointOptions, advanceToEndOfEventTime);

        return CompletableFuture.completedFuture(Acknowledge.get());
    } else {
        final String message = "TaskManager received a checkpoint request for unknown task " + executionAttemptID + '.';

        log.debug(message);
        return FutureUtils.completedExceptionally(new CheckpointException(message, CheckpointFailureReason.TASK_CHECKPOINT_FAILURE));
    }
}

Flink Checkpoint 在 TaskManager 端的處理過程是從 Source 開始的，JobManager 會給每個Source 算子發送一個 觸發 Checkpoint 的 Rpc 請求，TaskManager 接收到對應 Source 算子的 Checkpoint Rpc 請求后，就開始執行對應流程，同時會往自己的下游算子廣播 CheckpointBarrier。

對應 Kafka Source，執行 Checkpoint 的方法 是 FlinkKafkaConsumerBase.snapshotState，Checkpoint 的時候，從 TaskExecutor 到 FlinkKafkaConsumerBase.snapshotState 的調用棧如下，調用棧比較長，就簡單列下調用的方法

TaskExecutor.triggerCheckpoint  --> task.triggerCheckpointBarrier

Task.triggerCheckpointBarrier   --> invokable.triggerCheckpointAsync

SourceStreamTask.triggerCheckpointAsync   -->   super.triggerCheckpointAsync

StreamTask.triggerCheckpointAsync   -->  triggerCheckpoint

StreamTask.triggerCheckpoint  ---> performCheckpoint

StreamTask.performCheckpoint  --> subtaskCheckpointCoordinator.checkpointState

SubtaskCheckpointCoordinatorImpl.checkpointState       --->  takeSnapshotSync    (會廣播 CheckpointBarrier 到下游算子, takeSnapshotSync 成成功后會發送 ack 到 JobMaster)

SubtaskCheckpointCoordinatorImpl.takeSnapshotSync   --->  buildOperatorSnapshotFutures

SubtaskCheckpointCoordinatorImpl.buildOperatorSnapshotFutures   ---> checkpointStreamOperator

SubtaskCheckpointCoordinatorImpl.checkpointStreamOperator   --->  op.snapshotState

AbstractStreamOperator.snapshotState   ---> stateHandler.snapshotState

StreamOperatorStateHandler.snapshotState   --->  snapshotState              (會調用 operatorStateBackend.snapshot/keyedStateBackend.snapshot 將 state 持久化到 stateBackend )

StreamOperatorStateHandler.snapshotState   --->  streamOperator.snapshotState(snapshotContext)

AbstractUdfStreamOperator.snapshotState   ---> StreamingFunctionUtils.snapshotFunctionState

StreamingFunctionUtils.snapshotFunctionState   ---> trySnapshotFunctionState

StreamingFunctionUtils.trySnapshotFunctionState   ---> ((CheckpointedFunction) userFunction).snapshotState  (就進入 udf 了，這里是：FlinkKafkaConsumerBase， 如果是自定義的 Source，會進入對應Source 的 snapshotState 方法)

FlinkKafkaConsumerBase.snapshotState

Kafka Source Checkpoint 

Flink State 可以分為 KeyedState 和 OperatorState，KeyedState 在keyBy 算子之后使用，OperatorState 使用較多的就是存儲Source 和 Sink 的狀態，比如Kafka Source 存儲當前消費的 offset。 其他算子想使用 OperatorState 需要實現 CheckpointedFunction，Operator state 存在 taskManager 的 heap 中，不建議存儲大狀態。

Kafka Source 的 checkpoint 是在 FlinkKafkaConsumerBase 中實現的，具體方法是： snapshotState
FlinkKafkaConsumerBase 的 checkpoint 流程 大概是，獲取 kafkaFetcher 的

/** Accessor for state in the operator state backend. */
// ListState 存儲狀態，Checkpoint 的時候只需要將 KafkaTopicPartition 和 Offset 放入這個對象中，Checkpoint 的時候，就會寫入 statebackend
// Operator state 都是這樣的，自己實現的也是，將對應內容寫入狀態就可以了
private transient ListState<Tuple2<KafkaTopicPartition, Long>> unionOffsetStates;


@Override
public final void snapshotState(FunctionSnapshotContext context) throws Exception {
    if (!running) {
        LOG.debug("snapshotState() called on closed source");
    } else {
        // 消費者還在運行
        // 清楚之前的狀態
        unionOffsetStates.clear();

        final AbstractFetcher<?, ?> fetcher = this.kafkaFetcher;
        if (fetcher == null) {
            // the fetcher has not yet been initialized, which means we need to return the
            // originally restored offsets or the assigned partitions
            for (Map.Entry<KafkaTopicPartition, Long> subscribedPartition : subscribedPartitionsToStartOffsets.entrySet()) {
                unionOffsetStates.add(Tuple2.of(subscribedPartition.getKey(), subscribedPartition.getValue()));
            }

            if (offsetCommitMode == OffsetCommitMode.ON_CHECKPOINTS) {
                // the map cannot be asynchronously updated, because only one checkpoint call can happen
                // on this function at a time: either snapshotState() or notifyCheckpointComplete()
                pendingOffsetsToCommit.put(context.getCheckpointId(), restoredState);
            }
        } else {
            // 獲取對應的 KafkaTopicPartition 和 offset
            HashMap<KafkaTopicPartition, Long> currentOffsets = fetcher.snapshotCurrentState();
            // 放到 pendingOffsetsToCommit 中
            if (offsetCommitMode == OffsetCommitMode.ON_CHECKPOINTS) {
                // the map cannot be asynchronously updated, because only one checkpoint call can happen
                // on this function at a time: either snapshotState() or notifyCheckpointComplete()
                pendingOffsetsToCommit.put(context.getCheckpointId(), currentOffsets);
            }
            // 將 KafkaTopicPartition 和 offset 寫入 operator state 中
            for (Map.Entry<KafkaTopicPartition, Long> kafkaTopicPartitionLongEntry : currentOffsets.entrySet()) {
                unionOffsetStates.add( Tuple2.of(kafkaTopicPartitionLongEntry.getKey(), kafkaTopicPartitionLongEntry.getValue()));
            }
        }
        // 移除還未 提交的 offset
        if (offsetCommitMode == OffsetCommitMode.ON_CHECKPOINTS) {
            // truncate the map of pending offsets to commit, to prevent infinite growth
            while (pendingOffsetsToCommit.size() > MAX_NUM_PENDING_CHECKPOINTS) {
                pendingOffsetsToCommit.remove(0);
            }
        }
    }
}

當 Checkpoint 完成的時候，會調用到 FlinkKafkaConsumerBase 的 notifyCheckpointComplete 方法，會提交 offset 到 kafka 中，到這里 Kafka Source 的 Checkpoint 就完成了。

fetcher.commitInternalOffsetsToKafka(offsets, offsetCommitCallback);

算子快照完成后會給 JobMaster 發個消息說快照完成了

AsyncCheckpointRunnable.java

private void finishAndReportAsync(Map<OperatorID, OperatorSnapshotFutures> snapshotFutures, CheckpointMetaData metadata, CheckpointMetrics metrics, CheckpointOptions options) {
    // we are transferring ownership over snapshotInProgressList for cleanup to the thread, active on submit
    // AsyncCheckpointRunnable 的 run 方法，會給 JobMaster 發送 Checkpoint 完成的消息
    executorService.execute(new AsyncCheckpointRunnable(
        snapshotFutures,
        metadata,
        metrics,
        System.nanoTime(),
        taskName,
        registerConsumer(),
        unregisterConsumer(),
        env,
        asyncExceptionHandler));
}

AsyncCheckpointRunnable.run

public void run() {

    ..........
    if (asyncCheckpointState.compareAndSet(AsyncCheckpointState.RUNNING, AsyncCheckpointState.COMPLETED)) {
        // report ack
 reportCompletedSnapshotStates(
            jobManagerTaskOperatorSubtaskStates,
            localTaskOperatorSubtaskStates,
            asyncDurationMillis);

    } else {
        LOG.debug("{} - asynchronous part of checkpoint {} could not be completed because it was closed before.",
            taskName,
            checkpointMetaData.getCheckpointId());
    }
    .........

private void reportCompletedSnapshotStates(
    TaskStateSnapshot acknowledgedTaskStateSnapshot,
    TaskStateSnapshot localTaskStateSnapshot,
    long asyncDurationMillis) {


    // we signal stateless tasks by reporting null, so that there are no attempts to assign empty state
    // to stateless tasks on restore. This enables simple job modifications that only concern
    // stateless without the need to assign them uids to match their (always empty) states.
    taskEnvironment.getTaskStateManager().reportTaskStateSnapshots(
        checkpointMetaData,
        checkpointMetrics,
        hasAckState ? acknowledgedTaskStateSnapshot : null,
        hasLocalState ? localTaskStateSnapshot : null);

}

一路往下查看，會找到發送 Rpc 消息的地方：

下游算子 map Checkpoint

看 map 的 Checkpoint 流程，直接在 Map 算子上打個斷點，看下 調用棧就知道了

StreamTaskNetworkInput.emitNext 處理輸入數據和消息（Checkpoint）

StreamTaskNetworkInput.processElement

OneInputStreamTask.emitRecord

StreamMap.processElement 調用 userFunction.map 就是 我們代碼中的map 了

由於 Map 沒有狀態需要緩存，所以沒有實現 CheckpointedFunction，這里只列 出 CheckpointBarrier 廣播部分

如果消息是 Checkpoint：

StreamTaskNetworkInput.emitNext

@Override
public InputStatus emitNext(DataOutput<T> output) throws Exception {

    while (true) {
        // get the stream element from the deserializer
        if (currentRecordDeserializer != null) {
            DeserializationResult result = currentRecordDeserializer.getNextRecord(deserializationDelegate);
            if (result.isBufferConsumed()) {
                currentRecordDeserializer.getCurrentBuffer().recycleBuffer();
                currentRecordDeserializer = null;
            }

            if (result.isFullRecord()) {
                // 數據的處理流程，從這里調用直到 user function 的map 中
                processElement(deserializationDelegate.getInstance(), output);
                return InputStatus.MORE_AVAILABLE;
            }
        }
        // 從 Checkpoint InputGate 讀 CheckpointBarrier
        Optional<BufferOrEvent> bufferOrEvent = checkpointedInputGate.pollNext();
        ...
    }
}

CheckpointedInputGate.pollNext

@Override
public Optional<BufferOrEvent> pollNext() throws Exception {
    while (true) {

        ....
        Optional<BufferOrEvent> next = inputGate.pollNext();

        else if (bufferOrEvent.getEvent().getClass() == CheckpointBarrier.class) {
            CheckpointBarrier checkpointBarrier = (CheckpointBarrier) bufferOrEvent.getEvent();
            // CheckpointBarrier 的處理流程
            barrierHandler.processBarrier(checkpointBarrier, bufferOrEvent.getChannelInfo());
            return next;
        }
    ...
    }
}

barrierHandler.processBarrier 方法中對 Checkpoint 的處理流程跟 FlinkKafkaConsumerBase.snapshotState 調用的流程差不多

從 barrierHandler.processBarrier 調用到 SubtaskCheckpointCoordinatorImpl.checkpointState 往下游廣播 CheckpointBarrier

Kafka Sink Checkpoint

Kafka Sink 的 Checkpoint 也是從 Sink 收到 CheckpointBarrier 開始的

接收 CheckpointBarrier 的流程和 Map 一樣（所有算子都一樣，Source 是生成 CheckpointBarrier 的算子）

之后的流程就和 Source 一樣，一路調用到 FlinkKafkaConsumerBase.snapshotState 做快照

與 Kafka Source 一樣，Kafka Sink 也是將這次提交的內容放入 ListState 中，Sink 的 Checkpoint 實現了 TwoPhaseCommitSinkFunction（用以實現 精確一次 語義）

FlinkKafkaProducer.java

/**
 * State for nextTransactionalIdHint.
 */
private transient ListState<FlinkKafkaProducer.NextTransactionalIdHint> nextTransactionalIdHintState;

@Override
public void snapshotState(FunctionSnapshotContext context) throws Exception {
    super.snapshotState(context);

    nextTransactionalIdHintState.clear();
    // To avoid duplication only first subtask keeps track of next transactional id hint. Otherwise all of the
    // subtasks would write exactly same information.
    if (getRuntimeContext().getIndexOfThisSubtask() == 0 && semantic == FlinkKafkaProducer.Semantic.EXACTLY_ONCE) {
        checkState(nextTransactionalIdHint != null, "nextTransactionalIdHint must be set for EXACTLY_ONCE");
        long nextFreeTransactionalId = nextTransactionalIdHint.nextFreeTransactionalId;

        // If we scaled up, some (unknown) subtask must have created new transactional ids from scratch. In that
        // case we adjust nextFreeTransactionalId by the range of transactionalIds that could be used for this
        // scaling up.
        if (getRuntimeContext().getNumberOfParallelSubtasks() > nextTransactionalIdHint.lastParallelism) {
            nextFreeTransactionalId += getRuntimeContext().getNumberOfParallelSubtasks() * kafkaProducersPoolSize;
        }
        // 精確一次語義的 Checkpoint 狀態
        nextTransactionalIdHintState.add(new FlinkKafkaProducer.NextTransactionalIdHint(
            getRuntimeContext().getNumberOfParallelSubtasks(),
            nextFreeTransactionalId));
    }
}

TwoPhaseCommitSinkFunction.java

protected transient ListState<State<TXN, CONTEXT>> state;

@Override
public void snapshotState(FunctionSnapshotContext context) throws Exception {
    // this is like the pre-commit of a 2-phase-commit transaction
    // we are ready to commit and remember the transaction

    checkState(currentTransactionHolder != null, "bug: no transaction object when performing state snapshot");

    long checkpointId = context.getCheckpointId();
    LOG.debug("{} - checkpoint {} triggered, flushing transaction '{}'", name(), context.getCheckpointId(), currentTransactionHolder);
    // 預提交即調用 producer.flush 提交數據到 Kafka
    preCommit(currentTransactionHolder.handle);
    pendingCommitTransactions.put(checkpointId, currentTransactionHolder);
    LOG.debug("{} - stored pending transactions {}", name(), pendingCommitTransactions);
    //     開啟事務
    currentTransactionHolder = beginTransactionInternal();
    LOG.debug("{} - started new transaction '{}'", name(), currentTransactionHolder);

    state.clear();
    // 將 事務信息寫入 狀態
    state.add(new State<>(
        this.currentTransactionHolder,
        new ArrayList<>(pendingCommitTransactions.values()), userContext));
}

看到這里，應該都發現了 FlinkkafkaConsumerBase 和 TwoPhaseCommitSinkFunction 都有 notifyCheckpointComplete，在這個方法才真正完成 Checkpoint 往外部數據寫入 offset，提交事務。

注： Sink 完成 snaposhot 完成后會給 JobMaster 發送 ack 消息，與 Source 部分相同

JobManager 發送 confirmCheckpoint 消息給 TaskManager

JobManager 接收 checkpoint snapshotState 完成的消息

jobmanager接收完成 snapshotState 消息，然后會給 TaskManager 發送 所以算子完成快照的消息，調用算子的 notifyCheckpointComplete 方法，完成 Checkpoint 全部過程。

CheckpointCoordinator.receiveAcknowledgeMessage

CheckpointCoordinator.completePendingCheckpoint

CheckpointCoordinator.sendAcknowledgeMessages

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.checkpointComplete(checkpointId);
    }
}

從 ee.notifyCheckpointComplete 進去，可以看到發送 Rpc 消息的地方

Kafka Source/Sink 接收 notifyCheckpointComplete

對於 Source 從TaskManager 收到 confirmCheckpoint 開始

圖片： tm接收confirmCheckpoint消息

一路調用到 FlinkKafkaConsumerBase.notifyCheckpointComplete 提交 offset 到 Kafka

Sink 基本一樣，不過 FlinkKafkaProducer 的 notifyCheckpointComplete 在 TwoPhaseCommitSinkFunction 中（繼承來的）

里面會調用到 FlinkKafkaProducer.commit 方法，提交 Kafka 事務

@Override
protected void commit(FlinkKafkaProducer.KafkaTransactionState transaction) {
    if (transaction.isTransactional()) {
        try {
            // 調用 KafkaProduce 的方法，提交事務
            transaction.producer.commitTransaction();
        } finally {
            recycleTransactionalProducer(transaction.producer);
        }
    }
}

至此 Checkpoint的前台流程就全部完成了

 

歡迎關注Flink菜鳥公眾號，會不定期更新Flink（開發技術）相關的推文