聊聊kafka的group coordinator

forrestou 2019-06-25

本文主要来讲一个kafka的group coordinator。在kafka0.9.0版本的时候,开始启用了新的consumer config,这个新的consumer config采用bootstrap.servers替代之前版本的zookeeper.connect,主要是要渐渐弱化zk的依赖,把zk依赖隐藏到broker背后。

group coordinator

使用bootstrap.servers替代之前版本的zookeeper.connect,相关的有如下两个改动:

  • 在 Server 端增加了 GroupCoordinator 这个角色
  • 将 topic 的 offset 信息由之前存储在 zookeeper(/consumers/<group.id>/offsets/<topic>/<partitionId>,zk写操作性能不高) 上改为存储到一个特殊的 topic 中(__consumer_offsets)

从0.8.2版本开始Kafka开始支持将consumer的位移信息保存在Kafka内部的topic中(从0.9.0版本开始默认将offset存储到系统topic中)

Coordinator一般指的是运行在broker上的group Coordinator,用于管理Consumer Group中各个成员,每个KafkaServer都有一个GroupCoordinator实例,管理多个消费者组,主要用于offset位移管理和Consumer Rebalance。

rebalance时机

在如下条件下,partition要在consumer中重新分配:

  • 条件1:有新的consumer加入
  • 条件2:旧的consumer挂了
  • 条件3:coordinator挂了,集群选举出新的coordinator
  • 条件4:topic的partition新加
  • 条件5:consumer调用unsubscrible(),取消topic的订阅

__consumer_offsets

Consumer通过发送OffsetCommitRequest请求到指定broker(偏移量管理者)提交偏移量。这个请求中包含一系列分区以及在这些分区中的消费位置(偏移量)。偏移量管理者会追加键值(key-value)形式的消息到一个指定的topic(__consumer_offsets)。key是由consumerGroup-topic-partition组成的,而value是偏移量。
聊聊kafka的group coordinator

内存中也会维护一份最近的记录,为了在指定key的情况下能快速的给出OffsetFetchRequests而不用扫描全部偏移量topic日志。如果偏移量管理者因某种原因失败,新的broker将会成为偏移量管理者并且通过扫描偏移量topic来重新生成偏移量缓存。
聊聊kafka的group coordinator

清除offset日志

配置

log.cleaner.enable=true

compact

聊聊kafka的group coordinator

doc

相关推荐