问题现象

9月7号早上6点07分左右,线上有3个服务出现了异常提醒,kafka都提示了相同的问题:
The coordinator is not aware of this member.
从日志上看出,在出现该异常出现,kafka消费端与broker先出现了连接异常
在这里插入图片描述

问题分析

有多个服务多个topic出现了相同的异常,这里取其中一个topic研究异常发生的原因
从消费端日志:

[Consumer clientId=consumer-ygp-udc_goods_browse_record_group-2, groupId=ygp-udc_goods_browse_record_group] Offset commit failed on partition goods_browse_record-2 at offset 1162009: The coordinator is not aware of this member.

翻译过来大致的意思是:消费者组ygp-udc_goods_browse_record_group下的消费者consumer-ygp-udc_goods_browse_record_group-2在提交位移1162009到分区goods_browse_record-2时提交失败,原因是coordinator不认识这个成员。

关于什么是coordinator可以参考:Kafka消费者组重平衡(一)

那么,coordinator为什么会不认识这个成员呢?

我们项目在生产环境使用的是腾讯云kafka
腾讯云技术人员提供了broker端的日志:

12023-09-07 06:06:55 383] INFO (GroupCoordinator 100645]: Preparing to rebalance group ygp
udc goods browse record group in state PreparingRebalance with old generation 179
consumer offsets-25) (reason: removing member consumer-ygp
udc goods browse record group-2/xxx.xxx.xx.432023-06-27 19:01:47:259-46bc3386-ab36-4404
a823-5be8b11e74c0 on heartbeat expiration) (kafka.coordinatorgroup.GroupCoordinator

从broker端日志可以看出,是由于consumer端心跳超时,导致发生了重平衡,将网络链接有异常的consumer踢出了消费者组。

用这个问题在网上查了一番,总结下来,大致有以下两种说法:

  • Consumer中出现单独的standalone consume,而且standalone consume的名字和某个GroupId名字相同,这种现象就会触发报错
  • 组成员“崩溃”,造成kafka被动Rebalance,这样就会触发The coordinator is not aware of this member

项目中并没有使用standalone consume;结合broker端日志,基本可以猜测:
consumer端遇到网络中断/阻塞,心跳会话超时,触发了Rebalance,consumer被踢出消费者组,网络恢复后,consumer再次提交位移,就出现了“不认识该成员”的异常。

3个重要的参数
当 Consumer Group 完成 Rebalance 之后,每个 Consumer 实例都会定期地向Coordinator 发送心跳请求,表明它还存活着。如果某个 Consumer 实例不能及时地发送心跳请求,Coordinator 就会认为该 Consumer 已经“死”了,从而将其从 Group中移除,然后开启新一轮 Rebalance。

session.timeout.ms : 默认值10s,如果 Coordinator 在 10 秒之内没有收到 Group 下某 Consumer 实例的心跳,它就会认为这个 Consumer 实例已经挂了。这个参数决定了Consumer 存活性的时间间隔。

heartbeat.interval.ms : 默认值3s,控制发送心跳请求频率的参数,每隔3s发送一次心跳。broker会在心跳请求的响应中返回是否需要开启重平衡。这个值设置得越小,Consumer 实例发送心跳请求的频率就越高。频繁地发送心跳请求会额外消耗带宽资源,但好处是能够更加快速地知晓当前是否开启Rebalance。

max.poll.interval.ms : Consumer 端应用程序两次调用 poll 方法的最大时间间隔。它的默认值是 5 分钟,表示你的 Consumer 程序如果在 5分钟之内无法消费完 poll 方法返回的消息,那么 Consumer 会主动发起“离开组”的请求,Coordinator 也会开启新一轮 Rebalance。

复现

猜想:consumer端遇到网络中断/阻塞,心跳会话超时,触发了Rebalance,consumer被踢出消费者组,网络恢复后,consumer再次提交位移,就出现了“不认识该成员”的异常。

这里的模拟有两个关键点

  • consumer跟broker端有出现了超过10s以上网络连接异常(session.timeout.ms决定的)
  • 出现Rebalance前,consumer端还有消费位移没有提交(这里通过模拟消息处理时长超过10s)

为验证猜想,消费端参数保持跟生产环境一致

spring:
  #kafka配置#
  kafka:
    listener:
      concurrency: 1
      type: single #默认值 single
      poll-timeout: PT30S
    bootstrap-servers: ${base.kafka.bootstrap-servers}
    consumer:
      auto-offset-reset: earliest # earliest,latest,none,exception;
      enable-auto-commit: true
      max-poll-records: 100
      fetch-min-size: 1
      fetch-max-wait: PT2S
      group-id: ${spring.application.name}

需要说明的是session.timeout.msheartbeat.interval.msmax.poll.interval.ms使用的是默认配置。

在kafka控制台创建一个topic(2个分区1个副本)

bin/kafka-topics.sh --create --bootstrap-server node1:9092 --replication-factor 1 --partitions 2 --topic test-rebalance
[root@node1 kafka_2.13-3.2.1]# bin/kafka-topics.sh --describe --bootstrap-server node1:9092 --topic balance-test
Topic: balance-test     TopicId: AVSLeHGSRLaE4elyUzqLcw PartitionCount: 2       ReplicationFactor: 1    Configs: segment.bytes=1073741824
        Topic: balance-test     Partition: 0    Leader: 0       Replicas: 0     Isr: 0
        Topic: balance-test     Partition: 1    Leader: 0       Replicas: 0     Isr: 0

同一组下,启动两个消费者,一个通过控制台消费,一个通过代码消费。

@KafkaListener(topics = "balance-test", groupId = "kafka-boot")
    public void testReBalance(ConsumerRecord<String, String> recordInfo) {
        int partition = recordInfo.partition();
        try {
            // session.timeout.ms默认时间是10,为了到达“已经触发了Rebalance,还有位移没有提交”的情况,
            // 这里sleep20s,模拟消息处理时长
            Thread.sleep(20000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("re balance Consumer partition:" + partition + " value:" + recordInfo.value());
    }

控制台消费端:

[root@node1 kafka_2.13-3.2.1]# bin/kafka-console-consumer.sh --bootstrap-server node1:9092 --consumer-property group.id=kafka-boot --topic balance-test

查看消费组kafka-boot的分布情况:


GROUP           TOPIC            PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID                                                HOST             CLIENT-ID
kafka-boot      balance-test     0          4               4               0               console-consumer-44dea1f0-5795-456e-86c0-33d4d1033acf      /127.0.0.1       console-consumer
kafka-boot      balance-test     1          3               3               0               consumer-kafka-boot-1-cb67b3ed-b441-4293-a4fa-29c0e4ebd1a6 /219.136.123.186 consumer-kafka-boot-1

在控制台发布消息,观察消费情况

控制台消费:
在这里插入图片描述
代码消费:
在这里插入图片描述
上面可以看出,代码跟控制台都能进行正常消费。

下面模拟异常情况。本地通过禁用网卡,断开与broker端的链接。大致10s后,重新链接网络,观察各个端的日志输出情况。

本地consumer代码端日志:
在这里插入图片描述
在这里插入图片描述
可以看到在提交位移10时,出现了The coordinator is not aware of this

再看服务端日志:
在这里插入图片描述
同样出现了生成环境提示的异常

Preparing to rebalance group kafka-boot in state PreparingRebalance with old generation 9 (__consumer_offsets-1) (reason: removing member consumer-kafka-boot-1-4cc86013-261d-41b8-9fc2-dcc050df6b1f on heartbeat expiration) (kafka.coordinator.group.GroupCoordinator)

小结

kafka出现The coordinator is not aware of this member异常的其中一个原因是:因为网络中断/阻塞,客户端心跳线程直接访问不了kafka服务器,导致consumer被踢出消费者组,恢复正常后提交位移,就会被认为该组员不存在

关于Rebalance会造成的影响参考:Kafka消费者组重平衡(一)

针对这次出现的线上问题,我们可以优化一下kafka的配置参数,从而减少rebalance出现的概率:

  1. 主要还是增加session和hearbeat的时间
spring.kafka.consumer.properties.session.timeout.ms=25000
spring.kafka.consumer.heartbeat-interval=9000

阿里云官方文档建议超时时间(session.timeout.ms)设置成 25s,最长不超过 30s。那么心跳间隔时间(heartbeat.interval.ms)就不超过 10s。
参考阿里文档:https://help.aliyun.com/knowledge_detail/154454.html

  1. 缩短业务处理消息的时间,比如使用线程池。
Logo

开放原子开发者工作坊旨在鼓励更多人参与开源活动,与志同道合的开发者们相互交流开发经验、分享开发心得、获取前沿技术趋势。工作坊有多种形式的开发者活动,如meetup、训练营等,主打技术交流,干货满满,真诚地邀请各位开发者共同参与!

更多推荐