Kafka 的消费组时不时就会断开连接

深藏阁楼爱情的钟 2022-09-10 13:08 12阅读 0赞

## 问题现象 ##

Kafka消费组时不时就会断开连接1，提示如下异常

2019-19-28 12:36:44.711 |-INFO  [kafka-coordinator-heartbeat-thread | loongshawn] org.apache.kafka.clients.consumer.internals.AbstractCoordinator [1234] -| [Consumer clientId=loongshawn-xxxxxxxxxxxxxxxxxxxxxxxxxxxxx-127.0.0.1, groupId=loongshawn] Member loongshawn-xxxxxxxxxxxxxxxxxxxxxxxxxxxxx-127.0.0.1 sending LeaveGroup request to coordinator xx.xx.xx.xx:9092 (id: 23456223346 rack: null) due to consumer poll timeout has expired. This means 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 processing messages. You can address this either by increasing max.poll.interval.ms or by reducing the maximum size o

> 2019-19-28 12:36:44.711 |-INFO \[kafka-coordinator-heartbeat-thread | loongshawn\] org.apache.kafka.clients.consumer.internals.AbstractCoordinator \[1234\] -| \[Consumer clientId=loongshawn-xxxxxxxxxxxxxxxxxxxxxxxxxxxxx-127.0.0.1, groupId=loongshawn\] Member loongshawn-xxxxxxxxxxxxxxxxxxxxxxxxxxxxx-127.0.0.1 sending LeaveGroup request to coordinator xx.xx.xx.xx:9092 (id: 23456223346 rack: null) due to consumer poll timeout has expired. This means 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 processing messages. You can address this either by increasing max.poll.interval.ms or by reducing the maximum size o

## 问题分析 ##

### 1、一个topic，多个consumer ###

2019-19-28 22:23:55 INFO AbstractCoordinator:654 - [Consumer clientId=consumer-9, groupId=track123_pc] Discovered group coordinator loongshawnserver1:9092 (id: 23456223346 rack: null)
    2019-19-28 22:23:55 INFO ConsumerCoordinator:458 - [Consumer clientId=consumer-9, groupId=track123_pc] Revoking previously assigned partitions []
    2019-19-28 22:23:55 INFO AbstractCoordinator:486 - [Consumer clientId=consumer-9, groupId=track123_pc] (Re-)joining group
    2019-19-28 22:23:57 INFO AbstractCoordinator:831 - [Consumer clientId=consumer-8, groupId=track123_pc] Attempt to heartbeat failed since group is rebalancing
    2019-19-28 22:24:01 INFO AbstractCoordinator:831 - [Consumer clientId=consumer-8, groupId=track123_pc] Attempt to heartbeat failed since group is rebalancing

> 2019-19-28 22:23:55 INFO AbstractCoordinator:654 - \[Consumer clientId=consumer-9, groupId=track123\_pc\] Discovered group coordinator loongshawnserver1:9092 (id: 23456223346 rack: null)  
> 2019-19-28 22:23:55 INFO ConsumerCoordinator:458 - \[Consumer clientId=consumer-9, groupId=track123\_pc\] Revoking previously assigned partitions \[\]  
> 2019-19-28 22:23:55 INFO AbstractCoordinator:486 - \[Consumer clientId=consumer-9, groupId=track123\_pc\] (Re-)joining group  
> 2019-19-28 22:23:57 INFO AbstractCoordinator:831 - \[Consumer clientId=consumer-8, groupId=track123\_pc\] Attempt to heartbeat failed since group is rebalancing  
> 2019-19-28 22:24:01 INFO AbstractCoordinator:831 - \[Consumer clientId=consumer-8, groupId=track123\_pc\] Attempt to heartbeat failed since group is rebalancing

*  代码端启动了二个consumer，导致二个consumer之间一直在rebalancing。
 *  解决办法，要么起一个consumer。如果不好控制consumer的启动个数，消费完了后就close掉。

即当消费速度过慢时有可能会触发rebalance2, 这批消息被分配到另一个消费者，然后新的消费者还会消费过慢，再次rebalance, 这样一直恶性循环下去。看到的现象即消费组时断时续。

### 2、consumer poll一次数据，处理时间过长，第二次consumer poll又来了 ###

2019-19-28  16:47:32 WARN AbstractCoordinator:1011 - [Consumer clientId=consumer-1, groupId=track123_pc] This member will leave the group because consumer poll timeout has expired. This means 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 processing messages. You can address this either by increasing max.poll.interval.ms or by reducing the maximum size of batches returned in poll() with max.poll.records.
    2019-19-28  16:47:32 INFO AbstractCoordinator:782 - [Consumer clientId=consumer-1, groupId=track123_pc] Sending LeaveGroup request to coordinator loongshawnserver1:9092 (id: 23456223346 rack: null)

> 2019-19-28 16:47:32 WARN AbstractCoordinator:1011 - \[Consumer clientId=consumer-1, groupId=track123\_pc\] This member will leave the group because consumer poll timeout has expired. This means 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 processing messages. You can address this either by increasing max.poll.interval.ms or by reducing the maximum size of batches returned in poll() with max.poll.records.  
> 2019-19-28 16:47:32 INFO AbstractCoordinator:782 - \[Consumer clientId=consumer-1, groupId=track123\_pc\] Sending LeaveGroup request to coordinator loongshawnserver1:9092 (id: 23456223346 rack: null)

*  上面的报错已经说的很清楚了，二次poll的时间间隔超过了max.poll.interval.ms设置。
 *  解决办法：1，加大max.poll.interval.ms的时间，2，减少max.poll.records的请求的条数，或者二个同时间调整。

## 解决办法 ##

依据如上分析调整kafka参数配置:

*  max.poll.records：100 (默认值 500)
 *  max.poll.interval.ms：600000 (默认值 300000，也就是5分钟)

这里要注意：怎么调整参数3，结合实际场景，自己多验证，才知道。可能有的环境默认参数OK，换个环境就不行。

不过总的来说，还是消费逻辑不简洁，建议从根本上解决问题，即持久化方案[《Kafka消息积压案例分析》][Kafka]。

--------------------

1.  [kafka系列七、kafka核心配置][kafka_kafka] ↩︎
2.  [使用Kafka时一定要注意防止消费速度过慢触发rebalance而导致的重复消费][Kafka_rebalance] ↩︎
3.  [解决 Kafka Consumer 卡顿的问题][Kafka Consumer] ↩︎

[Kafka]: https://blog.csdn.net/loongshawn/article/details/119388712
[kafka_kafka]: https://www.cnblogs.com/wangzhuxing/p/10111831.html#_label1_7
[Kafka_rebalance]: https://zhuanlan.zhihu.com/p/123361451
[Kafka Consumer]: http://luojinping.com/2017/11/12/%E8%A7%A3%E5%86%B3-Kafka-Consumer-%E5%8D%A1%E9%A1%BF%E7%9A%84%E9%97%AE%E9%A2%98/