Kafka消费者组静态成员（static consumer member)

秒速五厘米 2023-06-02 10:59 7阅读 0赞

Kafka 2.3发布后官网的Consumer参数中增加了一个新的参数：group.instance.id。下面是这个参数的解释：

> A unique identifier of the consumer instance provided by end user. Only non-empty strings are permitted. If set, the consumer is treated as a static member, which means that only one instance with this ID is allowed in the consumer group at any time. This can be used in combination with a larger session timeout to avoid group rebalances caused by transient unavailability (e.g. process restarts). If not set, the consumer will join the group as a dynamic member, which is the traditional behavior.

大致意思是：它是用户指定的一个consumer成员ID。每个消费者组下这些ID必须是唯一的。一旦设置了该ID，该消费者就会被视为是一个静态成员（Static Member）。静态成员配以较大的session超时设置能够避免因成员临时不可用（比如重启）而引发的Rebalance。由此可见，消费者组静态成员是2.3版本新引入的一个概念，主要是为了避免不必要的Rebalance。

### ** Rebalance Recap** ###

之前我们在[Kafka消费者组][Kafka]一文中讨论过Rebalance机制。它的主要作用是为消费者组下所有成员分配分区。Client端和Broker端需要同时参与到Rebalance过程。在Broker端，Coordinator组件负责处理成员管理，比如处理组成员发送的JoinGroup请求、SyncGroup请求、Heartbeat请求和LeaveGroup请求；在Client端，Leader Consumer成员接收Coordinator发送的成员订阅信息，然后根据一定的策略（Range/Round-Robin/Sticky/自定义）制定分配方案。

Rebalance发生的条件有三个：

*  成员数量发生变化，即有新成员加入或现有成员离组（包括主动离组和崩溃被动离组）
 *  订阅主题数量发生变化
 *  订阅主题分区数量发生变化

其实，后两个条件可以合并成一个，即Rebalance触发条件只有两个：1. 成员数量发生变化；2. 订阅信息发生变化。

Rebalance的流程在那篇文章中也谈到了：首先，各个成员发送JoinGroup请求入组，Coordinator会等待一段时间等它们加入——这段时间由所有成员中max.poll.interval.ms的最大值来决定（在Kafka Connect中则是有专属的参数rebalance.timeout.ms来指定）。之后各成员发送SyncGroup请求等待Coordinator发送分配方案，然后开始正常消费。在消费的同时，各个consumer还会定期（heartbeat.interval.ms）上报心跳，告诉Coordinator组件它还活着。

### ** Issues for Rebalance** ###

在实际场景中，因为成员离组而发生的Rebalance应该算是最多的，但有些场景下的Rebalance是非常不合理的。比如我们公司就有这样的痛点：Consumer的处理逻辑发生变更，必须要更新代码重新上线，此时就要引发Rebalance，但其实重启Consumer也许只需要几分钟而已，也就是说我的消费只要中断几分钟就可以了，Kafka完全没必要为这个就触发一轮Rebalance，更没有必要重新分配分区，维持之前的分配方案足矣。虽然社区提供的Sticky分配方案在一定程度上能够缓解此问题，但Rebalance的Stop The World(STW)的特性还是决定了生产环境中Rebalance越少越好。

###  Static Member ###

在目前的Rebalance设计中，消费者组下的每个实例都会被Coordinator分配一个成员ID，即member.id。很多Kafka用户都有过这样的疑问：我能手动设置这个member.id吗？很遗憾，这个memberID是Kafka自动生成的，在静态成员被引入前，规则是client.id-UUID，这里的client.id就是Consumer端参数client.id的值，而且这个ID会随着每轮Rebalance发生变化的。换句话说，Coordinator无法持久化地保存某个consumer实例的member.id。我想这可能是制约Rebalance时所有成员必须强制重新加入的部分原因，因为Coordinator无法记住每个成员都是谁。如果你看源代码，可以发现在每次Client重启回来发送JoinGroup时，它会封装一个UNKNOWN\_MEMBER\_ID的空串，没有任何有意义的信息给到Broker端。Coordinator接收到后只能把它当做是一个全新的成员。相反地，如果member.id能够被记住，那么Coordinator就可以容忍它短暂的离线而不开启Rebalance，从而缩短消费者组整体不可用的时间窗口。

为此，社区于2.3和2.4版本引入了静态成员（Static Member）的概念以及一个新的Consumer端参数：group.instance.id。一旦配置了该参数，成员将自动成为静态成员，否则的话和以前一样依然被视为是动态成员。你可以认为这个新参数是一个要被持久化的新member.id。它依然不能由用户指定，构建规则是\`group.instsance.id\`-UUID。和member.id不同的是，每次成员重启回来后，其静态成员ID值是不变的，因此之前分配给该成员的所有分区也是不变的，而且在没有超时前静态成员重启回来是不会触发Rebalance的。

###  静态成员Rebalance条件 ###

显然，静态成员触发Rebalance的难度要小于动态成员。如果使用了静态成员，现在触发Rebalance的条件变更为：

*  新成员加入组：这个条件依然不变。当有新成员加入时肯定会触发Rebalance重新分配分区
 *  Leader成员重新加入组：比如主题分配方案发生变更
 *  现有成员离组时间超过了session超时时间：即使它是静态成员，Coordinator也不会无限期地等待它。一旦超过了session超时时间依然会触发Rebalance
 *  Coordinator接收到LeaveGroup请求：成员主动通知Coordinator永久离组。毕竟Kafka还是要提供方法让一个成员能够永远地退出组，此时重启Rebalance还是必要的

###  请求协议变更 ###

为了支持group.instance.id，与消费者组相关的协议格式也要做对应的变化。我看了下官网，JoinGroup、SyncGroup、LeaveGroup和OffsetCommit请求的协议格式都做了相应的变更。比如JoinGroup请求的Request和Response格式都增加了group-instance-id字段，如下所示：

> JoinGroup Request (Version: 5) => group\_id session\_timeout\_ms rebalance\_timeout\_ms member\_id group\_instance\_id protocol\_type \[protocols\]  
> 　　group\_id => STRING  
> 　　session\_timeout\_ms => INT32  
> 　　rebalance\_timeout\_ms => INT32  
> 　　member\_id => STRING  
> 　　**group\_instance\_id => NULLABLE\_STRING**  
> 　　protocol\_type => STRING  
> 　　protocols => name metadata  
> 　　name => STRING  
> 　　metadata => BYTES
> 
> JoinGroup Response (Version: 5) => throttle\_time\_ms error\_code generation\_id protocol\_name leader member\_id \[members\]  
> 　　throttle\_time\_ms => INT32  
> 　　error\_code => INT16  
> 　　generation\_id => INT32  
> 　　protocol\_name => STRING  
> 　　leader => STRING  
> 　　member\_id => STRING  
> 　　members => member\_id group\_instance\_id metadata  
> 　　member\_id => STRING  
> 　　**group\_instance\_id => NULLABLE\_STRING**  
> 　　metadata => BYTES

其他请求格式的变更也是类似的，这里就不贴了。

###  其他变更 ###

鉴于目前静态成员短暂重启或不可用不会触发Rebalance的改动，社区对消费者组最大session过期时间也做了修改。之前Consumer端参数group.min.session.timeout.ms值是6秒——要想在这个时间内重启完一个应用通常都是很困难的，因此社区现在将该值默认值改为30分钟。这就是说，只要配置有静态成员的Consumer程序代码更新及重启在30分钟之内完成，Consumer Group就不会发生Rebalance。当然在这段时间内，该Consumer的消费进度会中断，但是分区分配方案不会发生变化。

###  总结 ###

目前静态成员的部分功能已经集成进Kafka 2.3版本，还有一部分功能正在开发中，未来会进到2.4版本中。从目前的设计来看，静态成员机制能够帮助我们规避很多线上环境中本不必要的Rebalance，应该说是个很令人期待的新特性。同时，社区针对Rebalance的Stop The World酝酿一次大的修正，即所谓的增量协同式Rebalance（Incremental Cooperative Rebalance）。大致思想是允许单个consumer实例自行采用增量或渐进式的方式进行Rebalance，避免全局的STW。相关的代码正在开发中，后续我也会带来这方面的功能介绍。

转载于:https://www.cnblogs.com/huxi2b/p/11386847.html

[Kafka]: https://www.cnblogs.com/huxi2b/p/6223228.html