我们已经知道，在CheckpointCoordinator中会触发数据源算子的Checkpoint操作，同时向下游节点发送CheckpointBarrier事件。当下游Task实例接收到上游节点发送的CheckpointBarrier事件消息，且接收到所有InputChannel中的CheckpointBarrier事件消息时，当前Task实例才会触发本节点的Checkpoint操作。
这样设定的目的是让下游节点将所有InputChannel中属于当前Checkpoint的数据全部接入本节点，然后再对数据元素进行处理，以保证数据的一致性，一旦出现异常也能从上一次Checkpoint持久化结果中恢复当前Task实例的状态数据。
一. CheckpointBarrier对齐过程
在CheckpointedInputGate.pollNext()方法中先从上游节点获取Buffer或Event类型数据，然后分别做相应的数据处理，pollNext()方法的主要逻辑如下。
- 获取cp的buffer和Event：
从
Optional<BufferOrEvent> next中获取BufferOrEvent类型数据，BufferOrEvent既定义了Buffer数据也定义了Event。
- 锁定等待Barrier事件：
如果当前InputChannel被barrierHandler对象锁定，则将所有的BufferOrEvent数据本地缓存，直到InputChannel的锁被打开。barrierHandler会等所有InputChannel的CheckpointBarrier事件消息全部到达节点后，才继续处理该Task实例的Buffer数据，保证数据计算结果的正确性。
- Barrier Reset
在Buffer数据的处理过程中，如果Buffer缓冲区被填满，会进行清理操作和Barrier Reset操作。
- 消息类型处理：
如果BufferOrEvent的消息类型为Buffer，则直接返回next；
如果是CheckpointBarrier类型，则接入的CheckpointBarrier事件，最终根据CheckpointBarrier对齐情况选择是否触发当前节点的Checkpoint操作。
如果接收到的是CancelCheckpointMarker事件，则取消本次Checkpoint操作。
如果接收到的是EndOfPartitionEvent事件，表示上游Partition中的数据已经消费完毕，此时调用barrierHandler.processEndOfPartition()方法进行处理，最后清理缓冲区中的Buffer数据。
// 获取BufferOrEvent数据
BufferOrEvent bufferOrEvent = next.get();
if (barrierHandler.isBlocked(offsetChannelIndex(bufferOrEvent.
getChannelIndex()))) {
// 如果当前channel被barrierHandler对象锁定，则将BufferOrEvent数据先缓存下来
bufferStorage.add(bufferOrEvent);
// 如果缓冲区被填满，则进行清理操作和Barrier Reset操作
if (bufferStorage.isFull()) {
barrierHandler.checkpointSizeLimitExceeded(bufferStorage.getMaxBufferedBytes());
bufferStorage.rollOver();
}
}else if (bufferOrEvent.isBuffer()) {
// 如果是业务数据则直接返回，留给算子处理
return next;
}else if (bufferOrEvent.getEvent().getClass() == CheckpointBarrier.class) {
// 如果是CheckpointBarrier类型的事件，则对接入的Barrier进行处理
CheckpointBarrier checkpointBarrier = (CheckpointBarrier) bufferOrEvent.getEvent();
if (!endOfInputGate) {
// 根据算子的对齐情况选择是否需要进行Checkpoint操作
if (barrierHandler.processBarrier(checkpointBarrier,
offsetChannelIndex(bufferOrEvent.
getChannelIndex()),
bufferStorage.getPendingBytes())) {
bufferStorage.rollOver();
}
}
}else if (bufferOrEvent.getEvent().getClass() == CancelCheckpointMarker.class) {
// 如果是CancelCheckpointMarker类型事件，则调用processCancellationBarrier()方
// 法进行处理
if (barrierHandler.processCancellationBarrier(
(CancelCheckpointMarker) bufferOrEvent.getEvent())) {
bufferStorage.rollOver();
}
}
通过以上步骤我们可以看出，在CheckpointedInputGate中实现了CheckpointBarrier对齐的全部过程，并通过bufferStorage对接入的Buffer数据进行缓存，直到CheckpointBarrier事件全部对齐，才会对接入的数据进行处理。
接下来我们重点看CheckpointBarrierHandler的具体实现
二. CheckpointBarrierHandler的具体实现
1. CheckpointBarrierHandler分类
CheckpointBarrier事件的对齐过程主要借助CheckpointBarrierHandler实现。
如图，CheckpointBarrierHandler主要有CheckpointBarrierAligner和CheckpointBarrierTracker两种子类实现。
- CheckpointBarrierAligner用于实现Exactly-Once数据的一致性保障，对所有InputChannel中的CheckpointBarrier进行严格的对齐控制，并决定Task实例中InputChannel的block和unblock时间点。
- CheckpointBarrierTracker则实现了At-Least-Once语义处理保障，并没有对CheckpointBarrier进行非常严格的控制。
2. CheckpointBarrierAligner实现Barrier对齐操作
在CheckpointInputGate中通过CheckpointBarrierHandler.processBarrier()方法处理接收到的CheckpointBarrier事件。
如代码所示，实现类CheckpointBarrierAligner.processBarrier()方法主要逻辑如下。
- 是否进行Barrier对齐操作
- 从receivedBarrier中获取barrierId，判断totalNumberOfInputChannels是否为1，如果InputChannel数量为1，则触发Checkpoint操作，不需要进行CheckpointBarrier对齐操作。
- 如果InputChannel数量不为1，则判断numBarriersReceived是否大于0，即是否已经开始接收CheckpointBarrier事件，并进行Barrier对齐操作。
- barrier具体操作：
- 如果barrierId == currentCheckpointId条件为True，则调用onBarrier()方法进行处理。
- 如果barrierId > currentCheckpointId，表明已经有新的Barrier事件发出，超过了当前的CheckpointId，这种情况就会忽略当前的Checkpoint，并调用beginNewAlignment()方法开启新的Checkpoint。
- 如果以上条件都不满足，表明当前的Checkpoint操作已经被取消或Barrier信息属于先前的Checkpoint操作，此时直接返回false。
- 满足numBarriersReceived + numClosedChannels ==
totalNumberOfInputChannels条件后，触发该节点的Checkpoint操作。实际上会调用notifyCheckpoint()方法触发该Task实例的Checkpoint操作。
public boolean processBarrier(CheckpointBarrier receivedBarrier,
int channelIndex,
long bufferedBytes) throws Exception {
// 首先获取barrierId
final long barrierId = receivedBarrier.getId();
// 如果InputChannels为1，直接触发Checkpoint操作，不需要对齐处理
if (totalNumberOfInputChannels == 1) {
if (barrierId > currentCheckpointId) {
// 提交新的Checkpoint操作
currentCheckpointId = barrierId;
notifyCheckpoint(receivedBarrier, bufferedBytes,
latestAlignmentDurationNanos);
}
return false;
}
boolean checkpointAborted = false;
if (numBarriersReceived > 0) {
// 继续进行对齐操作
if (barrierId == currentCheckpointId) {
onBarrier(channelIndex);
}else if (barrierId > currentCheckpointId) {
LOG.warn("{}: Received checkpoint barrier for checkpoint {} before
completing current checkpoint {}. " + "Skipping current checkpoint.",
taskName,
barrierId,
currentCheckpointId);
// 通知Task当前Checkpoint没有完成
notifyAbort(currentCheckpointId,
new CheckpointException(
"Barrier id: " + barrierId,
CheckpointFailureReason.CHECKPOINT_DECLINED_SUBSUMED));
// 终止当前的Checkpoint操作
releaseBlocksAndResetBarriers();
checkpointAborted = true;
// 开启新的Checkpoint操作
beginNewAlignment(barrierId, channelIndex);
}else {
return false;
}
}else if (barrierId > currentCheckpointId) {
// 创建新的Checkpoint
beginNewAlignment(barrierId, channelIndex);
}else {
return false;
}
// 当Barrier接收的数量加上Channel关闭的数量等于整个InputChannels的数量时触发
Checkpoint操作
if (numBarriersReceived + numClosedChannels == totalNumberOfInputChannels) {
if (LOG.isDebugEnabled()) {
LOG.debug("{}: Received all barriers, triggering checkpoint {} at {}.",
taskName,
receivedBarrier.getId(),
receivedBarrier.getTimestamp());
}
// 释放Block并重置Barrier
releaseBlocksAndResetBarriers();
// 开始触发Checkpoint操作
notifyCheckpoint(receivedBarrier, bufferedBytes,
latestAlignmentDurationNanos);
return true;
}
return checkpointAborted;
}
经过以上步骤，基本上完成了CheckpointBarrier的对齐操作，当CheckpointBarrier完成对齐操作后，接下来就是通过notifyCheckpoint()方法触发StreamTask节点的Checkpoint操作。
