转发请注明原创地址：http://www.cnblogs.com/dongxiao-yang/p/7700600.html
《flink-connector-kafka consumer的topic分区分配源码》一文提到了在flink-connector-kafka的consumer初始化的时候有三种offset提交模式：KAFKA_PERIODIC，DISABLED和ON_CHECKPOINTS。
其中ON_CHECKPOINTS表示在flink做完checkpoint后主动向kafka提交offset的方法，本文主要分析一下flink-connector-kafka在源码如何使用checkpoint机制实现offset的恢复和提交。
flink conusmer的实现基类FlinkKafkaConsumerBase定义如下，这个类实现了了与checkpoin相关的三个接口CheckpointedFunction，CheckpointedRestoring<HashMap<KafkaTopicPartition, Long>>，CheckpointListener。根据官网文档，CheckpointedRestoring的restoreState()方法已经被CheckpointedFunction的initializeState取代，所以重点关注三个方法实现
1initializeState() 实例初始化或者recover的时候调用
2snapshotState() 每次创建checkpoint的时候调用
3 notifyCheckpointComplete() 每次checkpoint结束的时候调用
public abstract class FlinkKafkaConsumerBase<T> extends RichParallelSourceFunction<T> implements CheckpointListener, ResultTypeQueryable<T>, CheckpointedFunction, CheckpointedRestoring<HashMap<KafkaTopicPartition, Long>> {
initializeState
@Override public final void initializeState(FunctionInitializationContext context) throws Exception { // we might have been restored via restoreState() which restores from legacy operator state if (!restored) { restored = context.isRestored(); } OperatorStateStore stateStore = context.getOperatorStateStore(); offsetsStateForCheckpoint = stateStore.getSerializableListState(DefaultOperatorStateBackend.DEFAULT_OPERATOR_STATE_NAME); if (context.isRestored()) { if (restoredState == null) { restoredState = new HashMap<>(); for (Tuple2<KafkaTopicPartition, Long> kafkaOffset : offsetsStateForCheckpoint.get()) { restoredState.put(kafkaOffset.f0, kafkaOffset.f1); } LOG.info("Setting restore state in the FlinkKafkaConsumer."); if (LOG.isDebugEnabled()) { LOG.debug("Using the following offsets: {}", restoredState); } } } else { LOG.info("No restore state for FlinkKafkaConsumer."); } }
这个方法的逻辑比较简单，在task恢复的时候从stateStore中序列化出来之前存储的ListState<Tuple2<KafkaTopicPartition, Long>> 状态数据，并放到restoredState这个变量，用于下面open方法直接恢复对应的分区和offset起始值。
snapshotState
@Override public final void snapshotState(FunctionSnapshotContext context) throws Exception { if (!running) { LOG.debug("snapshotState() called on closed source"); } else { offsetsStateForCheckpoint.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()) { offsetsStateForCheckpoint.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 { HashMap<KafkaTopicPartition, Long> currentOffsets = fetcher.snapshotCurrentState(); 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); } for (Map.Entry<KafkaTopicPartition, Long> kafkaTopicPartitionLongEntry : currentOffsets.entrySet()) { offsetsStateForCheckpoint.add( Tuple2.of(kafkaTopicPartitionLongEntry.getKey(), kafkaTopicPartitionLongEntry.getValue())); } } 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); } } } }
snapshot方法创建checkpoint的做法是把当前的KafkaTopicPartition和目前消费到的offset值不断存放到offsetsStateForCheckpoint这个state对象里，然后把当前的checkpointid和对应的offset存到pendingOffsetsToCommit这个linkmap。当前offset的获取分两个情况，初始化的时候（if (fetcher == null) {...}）和fetcher已经初始化成功，初始化的时候从restoredState获取，正常运行中获取fetcher.snapshotCurrentState()。
notifyCheckpointComplete
public final void notifyCheckpointComplete(long checkpointId) throws Exception { if (!running) { LOG.debug("notifyCheckpointComplete() called on closed source"); return; } final AbstractFetcher<?, ?> fetcher = this.kafkaFetcher; if (fetcher == null) { LOG.debug("notifyCheckpointComplete() called on uninitialized source"); return; } if (offsetCommitMode == OffsetCommitMode.ON_CHECKPOINTS) { // only one commit operation must be in progress if (LOG.isDebugEnabled()) { LOG.debug("Committing offsets to Kafka/ZooKeeper for checkpoint " + checkpointId); } try { final int posInMap = pendingOffsetsToCommit.indexOf(checkpointId); if (posInMap == -1) { LOG.warn("Received confirmation for unknown checkpoint id {}", checkpointId); return; } @SuppressWarnings("unchecked") HashMap<KafkaTopicPartition, Long> offsets = (HashMap<KafkaTopicPartition, Long>) pendingOffsetsToCommit.remove(posInMap); // remove older checkpoints in map for (int i = 0; i < posInMap; i++) { pendingOffsetsToCommit.remove(0); } if (offsets == null || offsets.size() == 0) { LOG.debug("Checkpoint state was empty."); return; } fetcher.commitInternalOffsetsToKafka(offsets, offsetCommitCallback); } catch (Exception e) { if (running) { throw e; } // else ignore exception if we are no longer running } } }
notifyCheckpointComplete主要是在checkpoint结束后在ON_CHECKPOINTS的情况下向kafka集群commit offset，方法调用时会拿到已经完成的checkpointid，从前文的pendingOffsetsToCommit列表里找到对应的offset。如果判断索引不存在，则直接退出。否则，移除该索引对应的快照信息，然后将小于当前索引（较旧的）的快照信息也一并移除（这一点我之前解释过，因为所有的检查点都是按时间递增有序的）。最后将当前完成的检查点对应的消息的偏移量进行commit，也即commitOffsets。只不过这里该方法被定义为抽象方法，因为Kafka不同版本的API差别的原因，由适配不同版本的consumer各自实现，目前kafka09和010实现都是在Kafka09Fetcher内实现的commitInternalOffsetsToKafka方法。
参考文档：
http://blog.csdn.net/yanghua_kobe/article/details/51503885