8

After create multiple consumers (using Kafka 0.9 java API) and each thread started, I'm getting the following exception

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)

and then start consuming message normally, I would like to know what is causing this exception in order to fix it.

4
  • Hugo, are you still experiencing this issue? can you provide some more information?
    – Nautilus
    Mar 3, 2016 at 15:57
  • Yes @nautilus, I'm still having this issue. I have 3 consumers, all in the same consumer group, I have a topic with 20 partitions, from where data should be read. This exception occurs randomly, nevertheless consumers can read data from topic/partitions, although this exception is triggered. Mar 4, 2016 at 14:50
  • the consumers are only consuming the data or they are also processing it? I see on your stacktrace that the exception happens when you are trying to commitSync the offset, can you describe what happens between the consumption of the messages and the commit of the offset? I think that its possible that your consumer is losing heartbeat with the coordinator.
    – Nautilus
    Mar 4, 2016 at 23:08
  • Offsets have been committed after consume messages, but the exception still occurs. In other words, all messages are consumed no matter if the exceptions is triggered. Mar 7, 2016 at 22:19

2 Answers 2

11

Try also to tweak the following parameters:

  • heartbeat.interval.ms - This tells Kafka wait the specified amount of milliseconds before it consider the consumer will be considered "dead"
  • max.partition.fetch.bytes - This will limit the amount of messages (up to) the consumer will receive when polling.

I noticed that the rebalancing occurs if the consumer does not commit to Kafka before the heartbeat times out. If the commit occurs after the messages are processed, the amount of time to process them will determine these parameters. So, decreasing the number of messages and increasing the heartbeat time will help to avoid rebalancing.

Also consider to use more partitions, so there will be more threads processing your data, even with less messages per poll.

I wrote this small application to make tests. Hope it helps.

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

UPDATE

Kafka 0.10.x now offers a new parameter to control the number of messages received: - max.poll.records - The maximum number of records returned in a single call to poll().

UPDATE

Kafka offers a way to pause the queue. While the queue is paused, you can process the messages in a separated Thread, allowing you to call KafkaConsumer.poll() to send heartbeats. Then call KafkaConsumer.resume() after the processing is done. This way you mitigate the problems of causing rebalances due to not sending heartbeats. Here is an outline of what you can do :

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();
}
7
  • Version 0.10.x now has a new parameter, max.poll.records, to use instead of max.partition.fetch.bytes.
    – ajkret
    Sep 30, 2016 at 12:49
  • I used the same approach as yours of Pause and Resume, but still I get the same error. The only difference is I am calling commitSync() after pause() and before resume(), as I need to commit only if the records are processed. Any idea what I am doing wrong? Apr 8, 2017 at 8:18
  • @mav3n: I'm meeting the same issue. Did try increasing session.timeout.ms and max.poll.records but with no success. Have you found the method to do it? Jul 3, 2017 at 5:43
  • @HoàngLong : No, I didn't get any solution for that. Finally, I changed my design to a single consumer to make it work. Do share your findings if you get a solution to this. Thanks Jul 4, 2017 at 15:24
  • @mav3n: I have certain "findings" regarding this exception. First I increase session.timeout.ms on the Kafka server, there is no exception now but it is not optimal. Then we apply a "similar" way to ajkret's approach, except we don't use pause() and resume(), but periodically poll() the Kafka server every few seconds. Need to work out the interval, but it works for my case. Jul 5, 2017 at 1:51
0

Two possible reasons -->

  1. If there are any network failures, consumers cannot reach out to the broker and will throw this exception. But there were no network failures when these exceptions occurred.
  2. As mentioned in the error trace, if too much time is spent on processing the message, the ConsumerCoordinator will lose the connection and the commit will fail. This is because of polling.

The values given here are the default Kafka consumer configuration values.

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

Solution -->

Reduced the max.poll.records to 100 but still, the exception was occurring some times. So changed the configurations as below;

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

Reduced the heartbeat interval so that broker will be updated frequently that the Consumer is active. And also increased the session timeout configurations.

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.