项目场景：

项目中，使用到了kafka作为消息中间件，项目作为消费端，消费消息并进行业务处理

问题描述

在实际应用的过程中，发现偶尔但是一直存在的，有消费数据报：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.异常

原因分析：

首先根据报错提示，可能是max.poll.records参数和 max.poll.interval.ms参数配置问题
其次报错信息中提到消费者组正在发生重平衡导致

因此围绕这两道进行了大致分析：

参数配置没啥可说的，调整一下默认参数即可，这里没啥可说的
主要分析一下可能触发kafka重平衡的原因

重平衡

经过一番度娘，可能触发重平衡的原因如下：

主题的分区数发生变更，kafka目前只支持增加分区，当增加的时候就会触发重平衡
订阅的主题发生变化，当消费者组使用正则表达式订阅主题，而恰好又新建了对应的主题，就会触发重平衡
消费者组内成员发生变更，这个变更包括了增加和减少消费者。注意这里的减少有很大的可能是被动的，就是某个消费者崩溃退出了

首先考虑到前两种情况，出现的可能性都比较低，并且都是偶发性的，排除，因此主要针对第三点进行排查。实际排查过程中，是先修改了上面两个参数，发现错误并没有消失，因此在本地进行了错误模拟（业务处理部分直接sleep()），具体过程就不再赘述了，最终发现，本地消费者集群并且多线程消费时，当其中一个节点的某个消费者线程消费超时，会导致被剔除出消费者组，触发kafka的重平衡，此时同节点的其他消费线程和其他节点的消费者，正在消费的则有会报上面的异常，由此想到可能还是max.poll.interval.ms参数配置过小，然后对比线上日志，发现报错的消息，业务处理时长长达19分钟…

解决方案：

1、先将max.poll.interval.ms参数配置到大于19分钟，以避免线上继续出现问题
2、优化消费信息的业务处理逻辑，使之耗时不至于过长
3、最后再酌情配置max.poll.interval.ms参数

【记一次kafka报org.apache.kafka.clients.consumer.CommitFailedException异常处理】相关推荐

【Flink实战系列】Lorg/apache/flink/kafka/shaded/org/apache/kafka/clients/consumer/ConsumerRecord；)Ljava/
java.lang.AbstractMethodError: Method flink/stream/deserialization/PoJoDeserializationSchema.deseria ...
kafka 报错：kafka server：Message was too large ，server rejected it to avoid allocation
kafka报错信息: kafka server:Message was too large ,server rejected it to avoid allocation kafka版本: kafk ...
flink报错ByteArraySerializer is not an instance of org.apache.kafka.common.serialization.Serializer
文章目录问题描述报错信息问题解决方式一: 方式二: 问题描述代码如下: public class Sink_KafkaSink_1{public static void main(Strin ...
Flink读取Kafka报错：KafkaException ByteArrayDeserializer is not an instance Deserializer
1.视界 2.背景做flink读取kafka报错 org.apache.kafka.common.KafkaException: Failed to construct kafka consumer ...
kafka --org.apache.kafka.common.config.ConfigException: No resolvable bootstrap urls given in bootst
报错内容: Exception in thread "main" org.apache.kafka.common.KafkaException: Failed to constru ...
kafka grpc_模型服务：流处理与使用Java，gRPC，Apache Kafka，TensorFlow的RPC / REST
kafka grpc 机器学习/深度学习模型可以通过不同的方式进行预测. 我的首选方法是将分析模型直接部署到流处理应用程序(如Kafka Streams或KSQL )中. 您可以例如使用TensorF ...
模型服务：流处理与使用Java，gRPC，Apache Kafka，TensorFlow的RPC / REST
机器学习/深度学习模型可以通过不同的方式进行预测. 我的首选方法是将分析模型直接部署到流处理应用程序(如Kafka Streams或KSQL )中. 您可以例如使用TensorFlow for Jav ...
Spring Apache Kafka教程
在本SpringApache Kafka课程中,我们将学习如何在Spring Boot项目中开始使用Apache Kafka,并开始生成和使用我们所选主题的消息. 除了一个简单的项目外,我们还将深入探 ...
2.24. Spring boot with Apache Kafka
Spring boot 1.5.1 2.24.1. 安装 kafka 一下安装仅仅适合开发环境,生产环境请使用这个脚本安装 https://github.com/oscm/shell/tree/mas ...

【记一次kafka报org.apache.kafka.clients.consumer.CommitFailedException异常处理】

项目场景：

问题描述

原因分析：

解决方案：

【记一次kafka报org.apache.kafka.clients.consumer.CommitFailedException异常处理】相关推荐

最新文章

热门文章