checkpoint模式下offset的提交

在Flink kafka source源码解析(二)介绍了在没有开启checkpoint的时候，offset的提交方式，本文将重点介绍开启checkpoint后，flink kafka consumer提交offset的方式。

初始化offsetCommitMode

通过Flink kafka source源码解析(二)的介绍可以知道，当调用了env.enableCheckpointing方法后offsetCommitMode的值就是ON_CHECKPOINTS，而且会通过下面方法强制关闭kafka自动提交功能，这个值很重要，后续很多地方都是根据这个值去判断如果操作的。

  1. /**
  2.  * Make sure that auto commit is disabled when our offset commit mode is ON_CHECKPOINTS.
  3.  * This overwrites whatever setting the user configured in the properties.
  4.  * @param properties - Kafka configuration properties to be adjusted
  5.  * @param offsetCommitMode offset commit mode
  6.  */
  7. static void adjustAutoCommitConfig(Properties properties, OffsetCommitMode offsetCommitMode) {
  8.    if (offsetCommitMode == OffsetCommitMode.ON_CHECKPOINTS || offsetCommitMode == OffsetCommitMode.DISABLED) {
  9.       properties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
  10.    }
  11. }

保存offset

在做checkpoint的时候会调用FlinkKafkaConsumerBase#snapshotState方法，其中pendingOffsetsToCommit会保存要提交的offset

  1. if (offsetCommitMode == OffsetCommitMode.ON_CHECKPOINTS) {
  2.    // the map cannot be asynchronously updated, because only one checkpoint call can happen
  3.    // on this function at a time: either snapshotState() or notifyCheckpointComplete()
  4.    pendingOffsetsToCommit.put(context.getCheckpointId(), currentOffsets);
  5. }

同时，下面的变量会作为checkpoint的一部分保存下来，以便恢复时使用。

  1. /** Accessor for state in the operator state backend. */
  2. private transient ListState<Tuple2<KafkaTopicPartition, Long>> unionOffsetStates;

在snapshotState方法中会同时保存offset

  1. for (Map.Entry<KafkaTopicPartition, Long> subscribedPartition : subscribedPartitionsToStartOffsets.entrySet()) {
  2.     unionOffsetStates.add(Tuple2.of(subscribedPartition.getKey(), subscribedPartition.getValue()));
  3. }

提交offset

在checkpoint完成以后，task会调用notifyCheckpointComplete方法，里面判断offsetCommitMode == OffsetCommitMode.ON_CHECKPOINTS的时候，调用fetcher.commitInternalOffsetsToKafka(offsets, offsetCommitCallback);方法，最终会将要提交的offset通过KafkaFetcher#doCommitInternalOffsetsToKafka方法中的consumerThread.setOffsetsToCommit(offsetsToCommit, commitCallback);保存到KafkaConsumerThread.java中的nextOffsetsToCommit成员变量里面。

这样就会保证当有需要提交的offset的时候，下面代码会执行consumer.commitAsync，从而完成了手动提交offset到kafka。

  1. final Tuple2<Map<TopicPartition, OffsetAndMetadata>, KafkaCommitCallback> commitOffsetsAndCallback = nextOffsetsToCommit.getAndSet(null);
  3. if (commitOffsetsAndCallback != null) {
  4.   log.debug("Sending async offset commit request to Kafka broker");
  6.   // also record that a commit is already in progress
  7.   // the order here matters! first set the flag, then send the commit command.
  8.   commitInProgress = true;
  9.   consumer.commitAsync(commitOffsetsAndCallback.f0, new CommitCallback(commitOffsetsAndCallback.f1));
  10. }

总结

本文介绍了在checkpoint模式下，flink kafka source提交offset的方式，后续会介绍consumer读取offset的流程。

注：本文基于flink 1.9.0和kafka 2.3