kafka消费者程序日志报错Offset commit failed问题研究

生产环境偶尔会遇到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才对，这里就不再展开讨论了