关于flink的setCommitOffsetsOnCheckpoints

末蓝、 2022-11-10 05:35 33阅读 0赞

提交 offset 到 kafka,一般都是作为外部进度的监控,想实时知道作业消费的位置和 lag 情况

如果启用了 checkpoint，但是禁用 CommitOffsetsOnCheckpoints， kafka 消费组的 offset 不会提交到 kafka，也就是说： 消费组的 offset 是不会有变化的

也就是true or false设置后，flink都是主动管理。

如果是true，flink会把offset提交给kafka

用法示例:

//创建kafka数据流
    val properties = new Properties() properties.setProperty("bootstrap.servers", GlobalConfigUtils.getBootstrap) properties.setProperty("zookeeper.connect", GlobalConfigUtils.getZk) properties.setProperty("group.id", GlobalConfigUtils.getConsumerGroup) properties.setProperty("enable.auto.commit" , "true")//TODO properties.setProperty("auto.commit.interval.ms" , "5000") properties.setProperty("auto.offset.reset" , "latest") properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    val kafka09 = new FlinkKafkaConsumer09[String](
      GlobalConfigUtils.getIntputTopic,
      new SimpleStringSchema(),
      properties
    )
    /** *
    如果checkpoint启⽤用，当checkpoint完成之后，Flink Kafka Consumer将会提交offset保存 到checkpoint State中，
    这就保证了了kafka broker中的committed offset与 checkpoint stata中的offset相⼀一致。 ⽤用户可以在Consumer中调⽤用setCommitOffsetsOnCheckpoints(boolean) ⽅方法来选择启⽤用 或者禁⽤用offset committing(默认情况下是启⽤用的)
    * */
    kafka09.setCommitOffsetsOnCheckpoints(true)
    kafka09.setStartFromLatest()//start from the latest record
    kafka09.setStartFromGroupOffsets()
    //添加数据源addSource(kafka09)
    val data: DataStream[String] = env.addSource(kafka09)

**Reference:**

\[1\][【源码】 flink 消费 kafka 消费组 offset 提交][flink _ kafka _ offset]  
\[2\][flink⼿手动维护kafka偏移量量][flink_kafka]

[flink _ kafka _ offset]: https://www.cnblogs.com/Springmoon-venn/p/13405140.html
[flink_kafka]: https://my.oschina.net/u/4321334/blog/3955227