前言
前段时间写了三个Spark Streaming程序，负责从Kafka订阅群和用户消息，并做舆情监控必须的ETL工作。它们消费的Topic各自不同，但是分配的group.id都相同。运行一段时间之后，发现偶尔抛出如下所示的异常。
ERROR [org.apache.spark.streaming.scheduler.JobScheduler] - Error generating jobs for time 1579570560000 ms
java.lang.IllegalStateException: No current assignment for partition my_dummy_topic-11
at org.apache.kafka.clients.consumer.internals.SubscriptionState.assignedState(SubscriptionState.java:251)
at org.apache.kafka.clients.consumer.internals.SubscriptionState.needOffsetReset(SubscriptionState.java:315)
at org.apache.kafka.clients.consumer.KafkaConsumer.seekToEnd(KafkaConsumer.java:1170)
at org.apache.spark.streaming.kafka010.DirectKafkaInputDStream.latestOffsets(DirectKafkaInputDStream.scala:192)
at org.apache.spark.streaming.kafka010.DirectKafkaInputDStream.compute(DirectKafkaInputDStream.scala:209)
......
经过Google确定是由于我们使用的Spark Streaming 2.3版本无法处理Kafka Consumer Rebalance而引起的，见SPARK-22968及其对应的PR。我们通过进一步查看日志，确实发生了Rebalance。
那么本文就来说说Rebalance机制，先从执行Rebalance的主体——Consumer Group谈起。
Consumer Group
为了使得Consumer易于组织、可扩展以及更好地容错，Kafka将一个或多个Consumer组织为Consumer Group，即消费者组。Consumer Group的唯一标识就是group.id。Group内的所有Consumer共同消费已订阅的各个Topic的所有Partition，并且保证每个Partition只分配给该Group内的唯一一个Consumer。对Consumer Group的管理就是实现以下三个职责：
- 分配并维护Consumer与Partition的消费对应关系，即"who owns what"。（重新）分配消费对应关系的过程就叫Rebalance；
- 记录并提交Consumer消费Partition的消费位点（Offset值），即"consumed up to where"；
- 维护Consumer Group的配置信息与Consumer元数据。
下面简单介绍Consumer Group的管理。
Group Coordinator
在0.9版本之前，Kafka强依赖于ZooKeeper实现Consumer Group的管理：
- Group内每个Consumer通过在ZK内抢注节点来决定消费哪些Partition，并注册对Group和Broker相关节点的监听，以获知消费环境的变化（其他Consumer掉线、Broker宕机等），进而触发Rebalance；
- Offset值也维护在ZK中，老生常谈了。
这种方式除了过于依赖ZK，导致ZK压力偏大之外，还有两个分布式系统中常见且严重的问题：
- 羊群效应（herd effect）——一个被监听的ZK节点发生变化，导致大量的通知发送给所有监听者（即Consumer）；
- 脑裂（split brain）——ZK只保证最终一致性，不同的Consumer在同一时刻可能看到不同的Group和Broker状态，造成Rebalance混乱。
所以从0.9版本开始，Kafka重新设计了名为Group Coordinator的“协调者”服务负责实现上述职责，将这部分工作从ZK剥离开来。每个Broker在启动时，都会顺带启动一个Group Coordinator实例。每个Consumer Group在初始化时，都会分配给一个Group Coordinator实例来管理消费关系和Offset，如下简图所示。
Group Coordinator提交Offset时也不再是向ZK写，而是写入那个广为人知的特殊Topic——__consumer_offsets里。key是group-topic-partition格式的，value为Offset值。
那么该如何确定一个Consumer Group被分配给哪个Group Coordinator呢？Kafka根据groupId.hashCode() % offsets.topic.num.partitions取绝对值来得出该Group的Offset信息写入__consumer_offsets的分区号，并将Group分配给该分区Leader所在的Broker上的那个Group Coordinator。
上面粗略讲了Group Coordinator和它的第二个职责的实现方式，接下来仔细分析第一个职责，就是Rebalance的机制。
Rebalance触发条件与协议
我们已经知道，Rebalance就是一个Consumer Group内的所有Consumer分配消费各已订阅的Topic的各Partition的过程。Partition的分配有内置的三种算法实现（range、round-robin、sticky），用户也可以自定义分配规则，这个与本文关系不太大，就不再展开说。
那么有哪些条件会触发Rebalance呢？列举如下：
- Group发生变更，即新的Consumer加入，或原有的Consumer离开。Consumer离开可以是主动的（退出），也可以是被动的（崩溃）；
- Partition发生变更，包含订阅的Topic被创建或删除，以及现有订阅Topic新增Partition等。
本文只考虑Group发生变更的情况。Kafka定义了一个简单的协议来处理这种情况下的Rebalance过程，包含以下4对请求/响应：
- heartbeat：Group中的Consumer周期性地给Coordinator发送该心跳信号，表示自己存活；
- join-group：Consumer请求加入Group的信号；
- leave-group：Consumer主动退出Group的信号；
- sync-group：Coordinator将已生成的Consumer-Partition消费对应关系分发给Consumer的信号。
下面三张图示出心跳超时（被动退出）、主动退出和加入Group的情况。
图中的红色叹号表示Coordinator感知到了变化，并触发Rebalance过程。
Rebalance过程
整个Rebalance分为两个大步骤：JOIN和SYNC。
JOIN
正如其名，在这一步中，所有Consumer都会向Coordinator发送join-group，请求重新加入Group（那些原本已经在Group内的也不例外），同时放弃掉已分配给自己的Partition。以有新的Consumer主动加入Group（即上一节中的第三种情况）为例，图示如下。
图中的sync. barrier线就是表示JOIN步骤与SYNC步骤之间的分界，即从发起Rebalance开始到接受最后一个join-group之间的超时。需要特别注意的是，Broker端和Consumer端判定这个阶段超时的标准是不同的，Broker端使用rebalance.timeout.ms，而Consumer端使用max.poll.interval.ms。所以当遇到不明原因的长时间Rebalance时，应考虑max.poll.interval.ms的设定。
SYNC
这一步需要做的事情是：
- Coordinator在所有Consumer里选择一个担任Leader，并由Leader调用Partition分配规则来确定消费对应关系。
- 各个Consumer发送sync-group请求。Leader发送的请求里包含有已经确定的消费分配信息，其他Consumer的请求为空。
- Coordinator将消费分配信息原样封装在sync-group响应中，并投递给各个Consumer，最终使Group内所有成员都获知自己该消费的Partition。
图示如下，其中C1被选为Leader。SYNC步骤结束后，原有的1~6六个分区被分配给了C1~C3三个Consumer。Rebalance就完成了。
Rebalance的改进方案
上述Rebalance机制沿用了很长时间，但是它同样存在问题：
- stop-the-world问题——需要收回（revoke）所有Partition再重新分配（reassign），在此时间内，所有Consumer都无法进行消费。如果Rebalance时间长，会造成lag。
- back-and-forth问题——如果多次触发Rebalance，很有可能造成一个Consumer消费的Partition被分配给其他Consumer，然后又分配回来，做了无用功。
为了解决该问题，在2.4版本特别提出了Incremental（增量）Rebalance。顾名思义，就是Rebalance时不再让所有Consumer都放弃掉所有已分配的Partition，而是每次先记录，并转化成多次少量的Rebalance过程，且Consumer在此期间不会STW。简图如下所示。
关于该方案的细节，请参考KIP-429。
The End
民那晚安。祝身体健康，百毒不侵。
