Kafka CommitFailedException 消费者异常

Question

创建多个消费者（使用 Kafka 0.9 java API）并启动每个线程后，我得到以下异常

Consumer has failed with exception: org.apache.kafka.clients.consumer.CommitFailedException: Commit cannot be completed due to group rebalance
class com.messagehub.consumer.Consumer is shutting down.
org.apache.kafka.clients.consumer.CommitFailedException: Commit cannot be completed due to group rebalance
at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator$OffsetCommitResponseHandler.handle(ConsumerCoordinator.java:546)
at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator$OffsetCommitResponseHandler.handle(ConsumerCoordinator.java:487)
at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$CoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:681)
at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$CoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:654)
at org.apache.kafka.clients.consumer.internals.RequestFuture$1.onSuccess(RequestFuture.java:167)
at org.apache.kafka.clients.consumer.internals.RequestFuture.fireSuccess(RequestFuture.java:133)
at org.apache.kafka.clients.consumer.internals.RequestFuture.complete(RequestFuture.java:107)
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient$RequestFutureCompletionHandler.onComplete(ConsumerNetworkClient.java:350)
at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:288)
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.clientPoll(ConsumerNetworkClient.java:303)
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:197)
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:187)
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:157)
at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.commitOffsetsSync(ConsumerCoordinator.java:352)
at org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:936)
at org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:905)

然后开始正常消费消息，我想知道是什么导致了这个异常以便修复它。

Answer 1

两个可能的原因 -->

1. 如果出现任何网络故障，消费者无法联系代理并会抛出此异常。但发生这些异常时并没有出现网络故障。
2. 如错误跟踪中所述，如果处理消息花费的时间过多，ConsumerCoordinator 将失去连接，提交将失败。这是因为轮询。

此处给出的值是默认的 Kafka 消费者配置值。

request.timeout.ms=40000
heartbeat.interval.ms=3000
max.poll.interval.ms=300000
max.poll.records=500
session.timeout.ms=10000

---

解决方案-->

将 max.poll.records 减少到 100 条但仍然，该异常有时会发生。所以改变了如下配置；

request.timeout.ms=300000
heartbeat.interval.ms=1000 
max.poll.interval.ms=900000
max.poll.records=100
session.timeout.ms=600000

减少了心跳间隔，以便代理会频繁更新消费者处于活动状态。并且还增加了会话超时配置。

Answer 2

同时尝试调整以下参数：

• heartbeat.interval.ms - 这告诉 Kafka 在考虑消费者将被视为“死亡”之前等待指定的毫秒数
• ma​​x.partition.fetch.bytes - 这将限制消费者在轮询时收到的消息数量（最多）。

我注意到如果消费者在心跳超时之前没有提交到 Kafka，就会发生重新平衡。如果在处理消息之后发生提交，则处理它们的时间将决定这些参数。因此，减少消息数量并增加心跳时间将有助于避免重新平衡。

还要考虑使用更多的分区，这样就会有更多的线程处理您的数据，即使每次轮询的消息更少。

我编写了这个小应用程序来进行测试。希望对您有所帮助。

https://github.com/ajkret/kafka-sample

更新

Kafka 0.10.x 现在提供了一个新参数来控制接收到的消息数量： - ma​​x.poll.records - 在一次 poll() 调用中返回的最大记录数。

更新

Kafka 提供了一种暂停队列的方法。当队列暂停时，您可以在单独的线程中处理消息，允许您调用 KafkaConsumer.poll() 来发送心跳。然后在处理完成后调用KafkaConsumer.resume()。通过这种方式，您可以缓解由于不发送心跳而导致重新平衡的问题。以下是您可以做什么的概述：

while(true) {
    ConsumerRecords records = consumer.poll(Integer.MAX_VALUE);
    consumer.commitSync();

    consumer.pause();
    for(ConsumerRecord record: records) {

        Future<Boolean> future = workers.submit(() -> {
            // Process
            return true;
        }); 


       while (true) {
            try {
                if (future.get(1, TimeUnit.SECONDS) != null) {
                    break;
                }
            } catch (java.util.concurrent.TimeoutException e) {
                getConsumer().poll(0);
            }
        }
    }

    consumer.resume();
}