The Exception you are referring to
Exception in thread "main" 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.
gives a hint on what is happening and what can be done to solve the problem. In the code this Exception is described as
“This exception is raised when an offset commit with KafkaConsumer#commitSync() fails with an unrecoverable error. This can happen when a group rebalance completes before the commit could be successfully applied. In this case, the commit cannot generally be retried because some of the partitions may have already been assigned to another member in the group.”
In my experience the thrown error message can be caused by different things although they are all related to the consumer not being assigned to the partition anymore:
- Creating more and more Consumers without closing them
- Timeout of poll
- Timeout of heartbeat
- Outdated Kerberos ticket
1. Opening more and more Consumers without closing them
A rebalance takes place if you add a consumer to an existing ConsumerGroup. Therefore, it is essential to close the consumer after usage or to always use the same instance instead of creating new KafkaConsumer object for every message/iteration.
2. Poll Timeout (as explained in the Error message):
[…] 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.
The configuration max.poll.interval.ms defaults to 300000ms or 5minutes. As your consumer is taking more than those 5 minutes, the consumer is considered failed and the group will rebalance in order to reassign the partitions to another member (see Consumer Configuration).
Solution for Poll Timeout:
A possible solution is also given in the error message
You can address this either by increasing
max.poll.interval.msor by reducing the maximum size of batches returned in poll() withmax.poll.records.
The consumer reads all messages again, because (as the error shows) it is not able to commit the offsets. That means, if you start the Consumer with the same group.id it think that it never read anything from that topic.
3. Timeout of Heartbeat
There are two main configuration in your KafkaConsumer that deal with the heartbeats: heartbeat.interval.ms and session.timeout.ms.
In a seperate background thread your KafkaConsumer sends periodic heartbeats to the server. If the consumer crashes or is unable to send heartbeats for a duration of session.timeout.ms, then the consumer will be considered dead and its partitions will be reassigned. If the rebalance is triggered your consumer can’t commit anything from an “old assigned” partition as it is written in the description of the CommitFailedException: “This can happen when a group rebalance completes before the commit could be successfully applied.”
Solution for Heartbeat Timeout:
Increase the settings heartbeat.interval.ms and session.timeout.ms while following the recommendation: ” The heartbeat.interval.ms must be set lower than session.timeout.ms, but typically should be set no higher than 1/3 of that value.”
Just keep in mind that changing these values always comes with a trade-off. You have either
- more frequent rebalances but shorter reaction time to identify dead consumers or
- less frequent rebalances and longer reaction time to identify dead consumers.
4. Outdated Kerberos ticket
On our production cluster we have seen the CommitFailedException just after the application was not able to renew the Kerberos ticket.