org.apache.kafka.clients.consumer.CommitFailedException

最近遇到一个kafka分区提异常如下：

throwable:org.apache.flink.kafka.shaded.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 max.poll.interval.ms or by reducing the maximum size of batches returned in poll() with max.poll.records.
at org.apache.flink.kafka.shaded.org.apache.kafka.clients.consumer.internals.ConsumerCoordinator$OffsetCommitResponseHandler.handle(ConsumerCoordinator.java:1151)
at org.apache.flink.kafka.shaded.org.apache.kafka.clients.consumer.internals.ConsumerCoordinator$OffsetCommitResponseHandler.handle(ConsumerCoordinator.java:1081)
at org.apache.flink.kafka.shaded.org.apache.kafka.clients.consumer.internals.AbstractCoordinator$CoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:1026)
at org.apache.flink.kafka.shaded.org.apache.kafka.clients.consumer.internals.AbstractCoordinator$CoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:1006)
at org.apache.flink.kafka.shaded.org.apache.kafka.clients.consumer.internals.RequestFuture$1.onSuccess(RequestFuture.java:204)
at org.apache.flink.kafka.shaded.org.apache.kafka.clients.consumer.internals.RequestFuture.fireSuccess(RequestFuture.java:167)
at org.apache.flink.kafka.shaded.org.apache.kafka.clients.consumer.internals.RequestFuture.complete(RequestFuture.java:127)
at org.apache.flink.kafka.shaded.org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient$RequestFutureCompletionHandler.fireCompletion(ConsumerNetworkClient.java:599)
at org.apache.flink.kafka.shaded.org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.firePendingCompletedRequests(ConsumerNetworkClient.java:409)
at org.apache.flink.kafka.shaded.org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:294)
at org.apache.flink.kafka.shaded.org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:233)
at org.apache.flink.kafka.shaded.org.apache.kafka.clients.consumer.KafkaConsumer.pollForFetches(KafkaConsumer.java:1300)
at org.apache.flink.kafka.shaded.org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1240)
at org.apache.flink.kafka.shaded.org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1168)
at org.apache.flink.streaming.connectors.kafka.internal.KafkaConsumerThread.run(KafkaConsumerThread.java:253)
message:Consumer subtask 80 failed async Kafka commit

从报错信息来看是说，有新的消费者加入发生了rebalance，导致offset提交失败。解决办法是增加max.poll.interval.ms或减少max.poll.records。
然而这次遇到的这个报错跟通常的提交失败绝对不一样，调整这两个参数根本不起作用。
具体什么原因，怎么解决请往下看。

背景

公司有一批spark streaming任务需要迁移到flink。为了保证数据的一致性，所以采取的方式是：flink任务复用了spark streaming任务的groupid，从而在spark任务停掉之后，flink能从kafka记录的offset接着消费。然而理想很丰满，在实际操作时候就报了上面的异常。
flink程序正常消费数据，就是无法提交offset到kafka，虽说flink自己也有Checkpoint来记录offset，但是一旦Checkpoint丢失，那么必定会造成数据的不准确。而且还有后续任务会根据kafka的offset来判断的逻辑。

原因

查阅了一些资料后，找到答案。原因是因为spark streaming消费kafka采用的是subscribe模式，而flink采用的是assign模式。同一个groupid不能接受两种模式的offset提交，具体原因情况这篇文章的分析。

解决方案

通过将旧groupid的offset信息复制到一个新的groupid上，然后flink用新groupid来消费即可。
具体复制使用python-kafka来实现，脚本如下：

from kafka import KafkaAdminClient, KafkaConsumerclient = KafkaAdminClient(bootstrap_servers='broker:9092')
meta = client.list_consumer_group_offsets('groupid_1')
consumer = KafkaConsumer(bootstrap_servers='broker:9092', group_id='groupid_2')
consumer.commit(meta)

org.apache.kafka.clients.consumer.CommitFailedException相关推荐

【记一次kafka报org.apache.kafka.clients.consumer.CommitFailedException异常处理】
项目场景: 项目中,使用到了kafka作为消息中间件,项目作为消费端,消费消息并进行业务处理问题描述在实际应用的过程中,发现偶尔但是一直存在的,有消费数据报:org.apache.kafka.cl ...
java.io.NotSerializableException: org.apache.kafka.clients.consumer.ConsumerRecord
java消费kafka数据时报错 ERROR [Executor task launch worker for task 90] - Exception in task 0.0 in stage 54 ...
【Flink实战系列】Lorg/apache/flink/kafka/shaded/org/apache/kafka/clients/consumer/ConsumerRecord；)Ljava/
java.lang.AbstractMethodError: Method flink/stream/deserialization/PoJoDeserializationSchema.deseria ...
java.lang.NoSuchMethodError: org.apache.kafka.clients.consumer.KafkaConsumer.assign(Ljava/util/List
版权声明:本文为博主原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接和本声明. 本文链接: Flink入门程序异常,记录一下跟大家分享. SLF4J: Failed to l ...
spark2+kafka报错:java.lang.NoSuchMethodError:org.apache.kafka.clients.consumer.KafkaConsumer.subscribe
spark读取kafka数据 // Create DataFrame representing the stream of input lines from kafkaval lines = spar ...
kafka生产者报错：[org.apache.kafka.clients.NetworkClient:600] - Error while fetching metadata with corre
在测试kafka写入数据的时候一直在报错: ... [org.apache.kafka.clients.NetworkClient:600] - Error while fetching meta ...
Error:(3, 41) java: 程序包org.apache.kafka.clients.producer不存在错误提示解决办法
解决 mvn idea:module
Apache Kafka Consumer 消费者集
1.目标在我们的上一篇文章中,我们讨论了Kafka Producer.今天,我们将讨论Kafka Consumer.首先,我们将看到什么是Kafka Consumer和Kafka Consumer的 ...
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 ...

org.apache.kafka.clients.consumer.CommitFailedException

背景

原因

解决方案

org.apache.kafka.clients.consumer.CommitFailedException相关推荐

最新文章

热门文章