记一次Kafka消费者频繁分区再均衡bug解决

Posted 2021-12-06 敲代码的小小酥

项目背景：

两个Kafka消费者群组，消费同样的主题，一个消费者群组消费数据后，发给mqtt服务，供其他应用接收;另一个消费者群组消费数据后，存入mysql数据库。

问题描述：

提示：这里描述项目中遇到的问题：
在项目中使用Kafka消费者消费数据，并配置了分区再均衡监听器，在日志文件中发现再均衡监听器频繁的输出日志，即频繁发生分区再均衡。而项目已经跑了很久，没有新的消费者加入消费者群组，为何还要频繁进行分区再均衡呢？
经过分析日志还发现，当发生再均衡时会输出如下信息:

Attempt to heartbeat failed since group is rebalancing

而且，偶尔还报如下错误:

org.apache.kafka.clients.consumer.CommitFailedException: Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member. This means that the time between subsequent calls to poll() was longer than the configured max.poll.interval.ms, which typically implies that the poll loop is spending too much time message processing. You can address this either by increasing the session timeout or by reducing the maximum size of batches returned in poll() with max.poll.records.

我们知道，频繁的进行分区再均衡极易偏移量提交不准确，造成数据重复消费等问题。那为何会发生如此现象呢？

---

原因分析：

提示：这里填写问题的分析：
经过进一步分析发现，输出上面两种日志信息的都是进行数据库入库的那个消费者群组，而实时转发到mqtt的那个消费者群组没有报这种错误。
而且从第二个报错信息来看，提示说两次poll的时间间隔超过了max.poll.interval.ms参数阈值，说明在一次poll中，处理数据的时间太长了，建议调大session timeout参数或者调小max.poll.records参数。
而第一个报错信息是说正在发生分区再均衡，心跳检测失败。
由此可以推测，是存入数据库的方法太占用时间了(一次poll会insert大量数据)，导致了频繁发生了再均衡，我们看mysql服务器的磁盘利用率，发现达到了100%，所以insert效率慢，造成了这个问题

---

解决方案：

提示：这里填写该问题的具体解决方案：
重新换了一个mysql服务器，再运行项目，问题解决。

知识积累：

这里了解一下Kafka消费者的心跳机制，以及Kafka剔除消费者的机制:
首先先要了解kafkaconsumer三个参数的区别:
max.poll.interval.ms、session.timeout.ms、heartbeat.interval.ms。
max.poll.interval.ms:当使用消费者群组时，一个消费者两次调用poll方法的最大延迟，如果超过了这个时间调用poll，则将会发生分区再均衡。
session.timeout.ms:配置消费者做大间隔发送心跳的时间阈值。如果超过这个阈值没发送心跳，则认为这个消费者挂了，会发生分区再均衡。group coordinator检测consumer发生崩溃所需的时间。
heartbeat.interval.ms: 消费者间隔多久向group coordinator发送一次心跳。消费者会周期性以这个值的参数进行心跳发送。

session.timeout.ms和heartbeat.interval.ms的区别是:
heartbeat.interval.ms设置消费者多久发送一次心跳，但是可能因为网络抖动等等原因，会偶尔的超过这个时间发送心跳或者少发送一次心跳。但是只要没超过session.timeout.ms的值发送心跳，那么就认为这个消费者是存活的。

特别注意的是，在早期版本，kafka的心跳机制和poll方法绑定的，也就是每poll一次，发送一次心跳。这种机制的缺陷就是当poll处理时间过久时，kafka认为该消费者挂掉了，直接移除出去了。所以在kafka1.0版本后，心跳和poll分成了两个线程:一个是heartbeat 线程，另一个是processing线程，processing线程可理解为调用consumer.poll方法执行消息处理逻辑的线程，而heartbeat线程是一个后台线程，对程序员是"隐藏不见"的。如果消息处理逻辑很复杂，比如说需要处理5min，那么 max.poll.interval.ms可设置成比5min大一点的值。而heartbeat 线程则和上面提到的参数 heartbeat.interval.ms有关，heartbeat线程 每隔heartbeat.interval.ms向coordinator发送一个心跳包，证明自己还活着。只要 heartbeat线程 在 session.timeout.ms 时间内 向 coordinator发送过心跳包，那么 group coordinator就认为当前的kafka consumer是活着的。

我们再分析上面的问题场景，两次poll间隔时间过长，超过了max.poll.interval.ms，所以导致了频繁的分区再均衡，而消费者的心跳是正常发的，所以每次再均衡，不会把处理慢的消费者移除出去，还会将其加入队列。以此频繁反复分区再均衡，其实质其实是mysql慢造成的分区再均衡。

参考文章:Kafka session.timeout.ms heartbeat.interval.ms参数的区别以及对数据存储的一些思考