生产环境偶尔会遇到kafka消费者程序日志报错的问题

截取主要日志如下:

2023-10-02 19:35:28.554 {trace: d7f97f70dd693e3d} ERROR[Thread-49:137] ConsumerCoordinator$OffsetCommitResponseHandler.handle(812) - [Consumer clientId=consumer-1, groupId=cid_yingzi_fpf_group_device] Offset commit failed on partition topic_dvc_telemetery_bh_bh100-1 at offset 4313614: The request timed out.

2023-10-02 19:35:28.554 {trace: d7f97f70dd693e3d} INFO [Thread-49:137] AbstractCoordinator.markCoordinatorUnknown(727) - [Consumer clientId=consumer-1, groupId=cid_yingzi_fpf_group_device] Group coordinator kafka02.yingzi.com:19292 (id: 2147483645 rack: null) is unavailable or invalid, will attempt rediscovery

kafka客户端版本为2.2.0

结合日志去阅读代码,只能大概定位到,是客户端程序向server发送commit offset请求的时候,server返回的错误信息是:The request timed out

看到 request timed out,第一时间很可能会误以为是客户端向server发送请求超时了。但是查看OffsetCommitResponseHandler.handle()的代码,发现server是成功返回信息了的。
server返回的数据是一个Map<TopicPartition, Errors>结构,每个partition都对应一个Errors结果,如果Errors为Errors.NONE,则表示offset提交成功。
如果Errors不为Errors.NONE,则打印错误日志,也就是上面的 Offset commit failed … The request timed out的日志,每个partition打印1条日志。
也就是说,问题发生在server内部处理的时候,可以排除掉是客户端和server的网络问题导致的超时

要继续深挖,需要了解下server的处理逻辑,server的入口代码在KafkaApis.handleOffsetCommitRequest()
查看代码逻辑,可以发现早期的offset是保存在zk中,新版本中改为存在kafka的topic中(往__consumer_offsets这个topic发消息,每个partition一条offset消息)
那么分析下来,大概率就是往__consumer_offsets topic发消息的时候,产生了超时

继续阅读client的代码,了解Offset commit的机制
在KafkaConsumer.poll()的代码中,每次拉取消息时,都会调用updateAssignmentMetadataIfNeeded()这个方法,这个方法最终会调用maybeAutoCommitOffsetsAsync()方法
maybeAutoCommitOffsetsAsync()方法根据autoCommitIntervalMs来判断,是否要提交offset
这里默认是5秒执行一次commit offset请求,每次会把订阅的所有topic和partition的信息都进行提交
每个topic每个partition对应1条消息,如果topic非常多的话,那往__consumer_offsets发送消息量也会很大

查看生产kafka的监控,__consumer_offsets每秒消息量大概为四五千
显然是不太合理的
于是通过命令去消费__consumer_offsets的数据进行查看,注意由于这里消息是序列化的,直接消费的话会显示乱码,要通过-formatter "kafka.coordinator.group.GroupMetadataManager$OffsetsMessageFormatter"去进行消息解码

测试环境消费命令如下:

/opt/software/kafka/kafka_2.11-2.0.0_test_yewu/bin/kafka-console-consumer.sh --topic __consumer_offsets --group yz.bigdata.tzy --bootstrap-server kafka-test01.yingzi.com:32295 --formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" --consumer.config config/consumer.properties

生产环境统计的消息量最多的group如下(消费25秒):

37485 cid_opf_scene_edgengine_offline_alarm
27216 idp_share_telemetery_scene
25776 idp_scene_lifecycle_offline_alarm
4892 cid_scene_dvc_tele_attr
1440 yingzi-scene-space-group
546 clickhouse-iotcenter-latest-rep
533 re-Main-consumer-b
533 clickhouse-iotcenter-rep
437 yingzi-bizcenter-dvc-telemetery-group
318 cid_yingzi_hwm_group
288 cid_yingzi_fpf_group_device
270 transport-b-node-tb-mqtt-transport-ca-b-8
258 transport-b-node-tb-mqtt-transport-ca-b-6
222 transport-b-node-tb-mqtt-transport-ca-b-5
208 tb-core-b-consumer
200 yz.bigdata.wns

排名前几的都是跟设备有关的,消费几百个topic
按照上面的分析,每个group每秒发送的消息量应该为:1 / auto-commit-interval * Sum(topic partirionNum)
但是实际计算下来,感觉不应该这么高才对

先针对cid_opf_scene_edgengine_offline_alarm这个group进行查看,这个group订阅的topic有250个,每个topic 6个partition
1/52506=300
但是实际37485/25 = 1500
差了5倍之多

于是找到对应的开发人员,查看kafka的配置,发现配置:spring.kafka.consumer.auto-commit-interval=1000

提交offset的间隔默认5秒,被人工修改为1秒,正好相差5倍
其他两个消息量很高的group,经分析也是一样的问题

沟通后建议还是把spring.kafka.consumer.auto-commit-interval配置改回默认值,后续再继续进行观察

当然问题的根本原因其实还是设计不合理,kafka的性能本身就是会随着topic的增多而降低的,设计上应该尽量避免产生很多个topic才对,这里就不再展开讨论了

Logo

Kafka开源项目指南提供详尽教程,助开发者掌握其架构、配置和使用,实现高效数据流管理和实时处理。它高性能、可扩展,适合日志收集和实时数据处理,通过持久化保障数据安全,是企业大数据生态系统的核心。

更多推荐