Group Coordinator：Kafka 如何管理消费组

验证：Coordinator 是 abs(group.id.hashCode()) % 50 号 __consumer_offsets 分区的 Leader

命令 · shell $ kafka-consumer-groups.sh --describe --group coord-g1 --state # 显示 COORDINATOR $ kafka-topics.sh --describe --topic __consumer_offsets # 50 分区，3 副本，各分区 Leader # 用 Java String.hashCode 算出每个组名对应的分区号，再查该分区的 Leader

coord-g1: hashCode=-540632446  abs(hash)%50=46  __consumer_offsets-46 的 Leader=broker1(19192)  实测 COORDINATOR=19192  一致
coord-g2: hashCode=-540632445  abs(hash)%50=45  __consumer_offsets-45 的 Leader=broker3(19392)  实测 COORDINATOR=19392  一致
coord-g3: hashCode=-540632444  abs(hash)%50=44  __consumer_offsets-44 的 Leader=broker1(19192)  实测 COORDINATOR=19192  一致
coord-g4: hashCode=-540632443  abs(hash)%50=43  __consumer_offsets-43 的 Leader=broker3(19392)  实测 COORDINATOR=19392  一致
coord-g5: hashCode=-540632442  abs(hash)%50=42  __consumer_offsets-42 的 Leader=broker2(19292)  实测 COORDINATOR=19292  一致
coord-g6: hashCode=-540632441  abs(hash)%50=41  __consumer_offsets-41 的 Leader=broker3(19392)  实测 COORDINATOR=19392  一致

6 个消费组，用组名的哈希取模 50 得到分区号，该分区的 Leader 所在 Broker 就是 --state 查到的 Coordinator，6 个全部吻合。所以 Coordinator 不需要单独选举，谁是 __consumer_offsets 对应分区的 Leader，谁就是这个组的 Coordinator；这个 topic 的分区数就是 offsets.topic.num.partitions，这个集群里是 50，kafka-topics.sh --describe 显示 PartitionCount: 50, ReplicationFactor: 3。

验证：消费组提交的 offset，就存在这个分区里，别的分区没有

命令 · java // 三个新组 coordx-a/b/c 各读 7 条并 commitSync，预测分区分别是 21、20、19 // 用 OffsetsMessageFormatter 读取 __consumer_offsets 的 19、20、21、22 号分区

__consumer_offsets-19: 共 2 条; 其中 coordx-* 的 1 条
   group=coordx-c topic=poll-demo partition=0 提交的 offset=7
__consumer_offsets-20: 共 14587 条; 其中 coordx-* 的 1 条
   group=coordx-b topic=poll-demo partition=0 提交的 offset=7
__consumer_offsets-21: 共 1 条; 其中 coordx-* 的 1 条
   group=coordx-a topic=poll-demo partition=0 提交的 offset=7
__consumer_offsets-22: 共 0 条; 其中 coordx-* 的 0 条

每个组的提交记录都恰好出现在预测的分区里，不在别的分区。也就是说 Coordinator 管理的组状态和 offset 是持久化在这个内部 topic 的分区日志中的，并且随分区做 3 副本复制，所以换一个 Broker 当 Coordinator 时数据是现成的。20 号分区里有 14587 条记录，是之前实验遗留的其他消费组的提交，与本文无关。中间我还遇到一个坑：最初用 kafka-console-consumer.sh --group ... --max-messages 5 创建的组，--describe 显示没有提交任何 offset，__consumer_offsets 里也查不到记录（组虽然存在，Coordinator 也能查到，但没提交），后来改用 Java Consumer 显式 commitSync() 才有记录。

验证：Coordinator 所在 Broker 被杀，消费不重新分配分区，只是提交被卡住约 9 秒

命令 · java // topic coord-fail2: 1 分区，Leader 在 broker1；组 coordy-2 的 Coordinator 是 broker2 // Producer 每 10ms 一条；Consumer 每次 poll 拿到数据后 commitSync；第 9 秒 kill -9 broker2

共消费 1823 条，commitSync 成功 937 次，失败 0 次；最长消费间隔 9291ms（起点 9006ms，正是 kill 的时刻）
  [8539ms] assigned 1
  [9007ms] >>> kill -9 67339
  [30011ms] revoked 1            <- 只有我结束程序时的 revoke，中间没有 rebalance
kill 之后再查 coordinator = 127.0.0.1:19392（broker3，原来是 broker2）

这次数据分区的 Leader（broker1）没有受影响，被杀的只是 Coordinator（broker2）。结果：消费者没有经历任何 rebalance（没有 revoke、lost、assigned 事件），937 次 commitSync 全部成功，没有一次失败；最长的一次消费间隔是 9291 毫秒，起点就是 kill 的时刻。因为我的消费循环是"poll 一批、commitSync 一次"，提交要等新的 Coordinator 出现才能成功，所以整个循环被卡住了约 9 秒，量级与《Broker 崩溃后，副本如何接管 Leader》里查到的 broker.session.timeout.ms=9000 吻合（这只是量级吻合，我没有单独验证两者的因果）。之后 Coordinator 变成了 broker3——__consumer_offsets-7 分区 Leader 切换到了另一个副本。我没有对比"改成 commitAsync"时数据消费能不能不受影响，所以这里的"卡住 9 秒"是同步提交带来的，不能直接说成"消费本身停了 9 秒"。

结论

Group Coordinator 就是 abs(group.id.hashCode()) % 50 号 __consumer_offsets 分区的 Leader，组的成员状态和提交的 offset 都存放在这个分区的日志里，随分区复制。Coordinator 所在 Broker 宕机时，分区 Leader 切换即完成 Coordinator 切换，消费者不需要 rebalance，代价是同步提交在切换期间会被阻塞（这个实验里约 9 秒）。

分步演示：6个消费组的Coordinator，是怎么用哈希算出来的

每一步都是本文上面几节计算哈希、查询__consumer_offsets分区Leader得到的原始结果。