Kafka 0.8 Consumer Rebalance

1 Rebalance时机

0.10kafka的rebalance条件

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

当一个group中,有consumer加入或者离开时,会触发partitions均衡.均衡的最终目的,是提升topic的并发消费能力.

当consumer启动时,所触发的操作:

1. 首先进行"Consumer Id注册"，eg：/consumers/clientSearchBhvGp/ids/clientSearchBhvGp_yz4823.hadoop.data.sina.com.cn-1466668340717-ed3a5f41
2. 然后在"Consumer id"节点下注册一个watch用来监听当前group中其他consumer的"退出"和"加入";只要此znode path下节点列表变更,都会触发此group下consumer的负载均衡.(比如一个consumer失效,那么其他consumer接管partitions).
3. 在"Broker id"节点下,注册一个watch用来监听broker的存活情况;如果broker列表变更,将会触发所有的groups下的consumer重新balance.

2 消费Partition

Consumer Group中各个consumer是根据先后启动的顺序有序消费一个topic的所有partitions的。

分配算法

1. 假如topic1,具有如下partitions: P0,P1,P2,P3
2. 加入group中,有如下consumer: C0,C1
3. 首先根据partition索引号对partitions排序: P0,P1,P2,P3
4. 根据(consumer.id + '-'+ thread序号)排序: C0,C1
5. 计算倍数: M = [P0,P1,P2,P3].size / [C0,C1].size,本例值M=2(向上取整)
6. 然后依次分配partitions: C0 = [P0,P1],C1=[P2,P3],即Ci = [P(i * M),P((i + 1) * M -1)]

举例说明：80个partition，有12个线程，如何进行分配。

3 Consumer owner

zk路径：/consumers/[groupId]/owners/[topic]/[partitionId] -> consumerIdString + threadId索引编号

可以查看每个partition属于哪个Consumers线程。

4 Consumer offset

zk路径：/consumers/[groupId]/offsets/[topic]/[partitionId] -> long (offset)

这个是永久节点，用来跟踪每个consumer目前所消费的partition中最大的offset。

通过这个可以排查不正常的Consumers线程，如果它的Offset不增长，而该topic的其它partition的Offset在增加，说明该partition的消费是不正常的。

5 Rebalance的策略

Consumer Rebalance的控制策略是由每一个Consumer通过在Zookeeper上注册Watch完成的。每个Consumer被创建时会触发 Consumer Group的Rebalance，具体启动流程如下：

• High Level Consumer启动时将其ID注册到其Consumer Group下，在Zookeeper上的路径为/consumers/[consumer group]/ids/[consumer id]
• 在/consumers/[consumer group]/ids上注册Watch
• 在/brokers/ids上注册Watch
  如果Consumer通过Topic Filter创建消息流，则它会同时在/brokers/topics上也创建Watch
• 强制自己在其Consumer Group内启动Rebalance流程

该方式有如下缺陷：

• Herd effect：
  任何Broker或者Consumer的增减都会触发所有的Consumer的Rebalance

• Split Brain：
  每个Consumer分别单独通过Zookeeper判断哪些Broker和Consumer 宕机了，那么不同Consumer在同一时刻从Zookeeper“看”到的View就可能不一样，这是由Zookeeper的特性决定的，这就会造成不正确的Reblance尝试。

• 调整结果不可控：
  所有的Consumer都并不知道其它Consumer的Rebalance是否成功，这可能会导致Kafka工作在一个不正确的状态。（这个就是目前出现的bug）

根据Kafka社区wiki，Kafka作者正在考虑在还未发布的0.9.x版本中使用中心协调器(Coordinator)。大体思想是为所有Consumer Group的子集选举出一个Broker作为Coordinator，由它来管理Consumer的增减，然后生成Rebalance命令，并检查是否这些Rebalance。

6 影响Rebalance失败的因素

6.1 rebalance失败官方解释：

consumer rebalancing fails (you will see ConsumerRebalanceFailedException): This is due to conflicts when two consumers are trying to own the same topic partition. The log will show you what caused the conflict (search for “conflict in “).

If your consumer subscribes to many topics and your ZK server is busy, this could be caused by consumers not having enough time to see a consistent view of all consumers in the same group. If this is the case, try Increasing rebalance.max.retries and rebalance.backoff.ms.

Another reason could be that one of the consumers is hard killed. Other consumers during rebalancing won’t realize that consumer is gone after zookeeper.session.timeout.ms time. In the case, make sure that rebalance.max.retries * rebalance.backoff.ms > zookeeper.session.timeout.ms.

最后的这一条说，实际情况并不是如此：rebalance.max.retries(4) * rebalance.backoff.ms(2000) > zookeeper.session.timeout.ms(6000),集群原来是这样的配置，仍然出现了rebalance失败。

6.2 分析

1. rebalance 重试的sleep时间:kafka/consumer/ZookeeperConsumerConnector.scala:393

• "rebalance.backoff.ms","zookeeper.sync.time.ms", 2000

2. rebalance 重试次数超过4次，syncedRebalance抛出的是RuntimeException，在下面的代码过程中，将这个异常捕获了,只记录这个ERROR。

kafka/consumer/ZookeeperConsumerConnector.scala:328，正确的做法是捕获到RunTimeException异常，通过exit(-1)让JVM这个进程退出。对于OLS程序会让它，重启一个Container继续运行。

6.3 解决

• 加大重试时间:"rebalance.backoff.ms=5000"
• 加大retry: "rebalance.max.retries=10"
• 捕获"ConsumerRebalanceFailedException"，退出程序。

7 一个失败的例子

consumer group:client_search_hbv（12个线程）消费topic:weibo_common_act2（80个partition），出现6个partition无线程消费，表现的日志是Consumers Owner Instances is None。

1.现象

该6个partition本该被一个进程id：90ac 占用消费，zookeeper上有该ids的节点。查看90ac的启动日志，发现它rebalance 失败了4次。

2016-06-23 15:52:31,473 ERROR kafka.consumer.ZookeeperConsumerConnector: [clientSearchBhvGp_yz4834.hadoop.data.sina.com.cn-1466668272656-90a8bbdc], error during syncedRebalance
kafka.common.ConsumerRebalanceFailedException: clientSearchBhvGp_yz4834.hadoop.data.sina.com.cn-1466668272656-90a8bbdc can't rebalance after 4 retries
        at kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener.syncedRebalance(ZookeeperConsumerConnector.scala:397)
        at kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener$$anon$1.run(ZookeeperConsumerConnector.scala:326)

2.分析

44-49的partition没有被消费，由48.34的进程消费；48.34在争取44-49 partitions的过程中，节点的值没有被48.37的进程release，导致一直创建临时节点不成功。

1. 48.34

• 15:52:22,840 开始抢占partition 44 - 49
• 15:52:31,473 重试四次失败

2. 48.37

• 15:52:24,032 准备开始抢partition 50 - 55
• 15:53:05,034 释放Release partition 44-49 ownership

Consumers 实例 Rebalance的过程

• Stop fetcher，停止向原有的partition拉取数据
• 先释放自己已有的partition owner
• 去争取自己被分配的新的partition，若创建临时节点成功，则成功。
• 重新启动新的fetcher拉取数据

3.解决办法

为Kafka支持ols.kafka.property.zookeeper.session.timeout.ms=30000等属性。
在workflow.xml文件中，CfgString支持以 ols.kafka.property. 开头的属性.

用户修改程序的2个步骤

1. 修改pom.xml的OLS_Yarn依赖为 0.2.2.2

<dependency>
    <groupId>com.sina.mis</groupId>
    <artifactId>OLS_Yarn</artifactId>
    <version>0.2.2.2</version>
</dependency>

2. 提交的workflow.xml在添加

ols.kafka.property.rebalance.backoff.ms=5000,ols.kafka.property.rebalance.max.retries=10

8 总结

consumer rebalance失败是0.8版本的bug，在0.9以后，这个模块由组件Coordinator负责，能够保证rebalance成功。

从0.10.0-src来看，ZookeeperConsumerConnector已经重构了，新增了ConsumerCoordinator。