offset提交模式(非checkpoint)

消费kafka topic最为重要的部分就是对offset的管理，对于kafka提交offset的机制，可以参考kafka官方网。

而在flink kafka source中offset的提交模式有3种：

  1. public enum OffsetCommitMode {
  3.    /** Completely disable offset committing. */
  4.    DISABLED,
  6.    /** Commit offsets back to Kafka only when checkpoints are completed. */
  7.    ON_CHECKPOINTS,
  9.    /** Commit offsets periodically back to Kafka, using the auto commit functionality of internal Kafka clients. */
  10.    KAFKA_PERIODIC;
  11. }

初始化offsetCommitMode

在FlinkKafkaConsumerBase#open方法中初始化offsetCommitMode

  1. // determine the offset commit mode
  2. this.offsetCommitMode = OffsetCommitModes.fromConfiguration(
  3.                 getIsAutoCommitEnabled(),
  4.                 enableCommitOnCheckpoints,
  5.         ((StreamingRuntimeContext)getRuntimeContext()).isCheckpointingEnabled());
  • 方法getIsAutoCommitEnabled()的实现如下：
    1. protected boolean getIsAutoCommitEnabled() {
    2.  return getBoolean(properties, ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true) &&
    3.     PropertiesUtil.getLong(properties, ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 5000) > 0;
    4. }


也就是说只有enable.auto.commit=true并且auto.commit.interval.ms>0这个方法才会返回true

  • 变量enableCommitOnCheckpoints默认是true，可以调用setCommitOffsetsOnCheckpoints改变这个值
  • 当代码中调用了env.enableCheckpointing方法，isCheckpointingEnabled才会返回true

通过下面的代码返回真正的提交模式

  1. /**
  2.  * Determine the offset commit mode using several configuration values.
  3.  *
  4.  * @param enableAutoCommit whether or not auto committing is enabled in the provided Kafka properties.
  5.  * @param enableCommitOnCheckpoint whether or not committing on checkpoints is enabled.
  6.  * @param enableCheckpointing whether or not checkpoint is enabled for the consumer.
  7.  *
  8.  * @return the offset commit mode to use, based on the configuration values.
  9.  */
  10. public static OffsetCommitMode fromConfiguration(
  11.       boolean enableAutoCommit,
  12.       boolean enableCommitOnCheckpoint,
  13.       boolean enableCheckpointing) {
  15.    if (enableCheckpointing) {
  16.       // if checkpointing is enabled, the mode depends only on whether committing on checkpoints is enabled
  17.       return (enableCommitOnCheckpoint) ? OffsetCommitMode.ON_CHECKPOINTS : OffsetCommitMode.DISABLED;
  18.    } else {
  19.       // else, the mode depends only on whether auto committing is enabled in the provided Kafka properties
  20.       return (enableAutoCommit) ? OffsetCommitMode.KAFKA_PERIODIC : OffsetCommitMode.DISABLED;
  21.    }
  22. }

本文暂时不考虑checkpoint的场景，所以只考虑(enableAutoCommit) ? OffsetCommitMode.KAFKA_PERIODIC : OffsetCommitMode.DISABLED;

也就是如果客户端设置了enable.auto.commit=true那么就是KAFKA_PERIODIC，否则就是KAFKA_PERIODIC。

offset的提交

自动提交

这种方式完全依靠kafka自身的特性进行提交，如下方式指定参数即可

  1. Properties properties = new Properties();
  2. properties.put("enable.auto.commit", "true");
  3. properties.setProperty("auto.commit.interval.ms", "1000");
  4. new FlinkKafkaConsumer<>("foo", new KafkaEventSchema(), properties)

非自动提交

通过上面的分析，如果enable.auto.commit=false，那么offsetCommitMode就是DISABLED
kafka官方文档中，提到当enable.auto.commit=false时候需要手动提交offset，也就是需要调用consumer.commitSync();方法提交。
但是在flink中，非checkpoint模式下，不会调用consumer.commitSync();，一旦关闭自动提交，意味着kafka不知道当前的consumer group每次消费到了哪。
可以从两方面证实这个问题：

  • 源码
    KafkaConsumerThread#run方法中是有consumer.commitSync();，但是只有当commitOffsetsAndCallback != null的时候才会调用。只有开启了checkpoint功能才会不为null，这个变量会在后续的文章中详细分析。
  • 测试
    1. 可以通过消费__consumer_offsets观察是否有offset的提交
    2. 重启程序，还是会重复消费之前消费过的数据

总结

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

注：本文基于flink 1.9.0和kafka 2.3