Problems committing offsets during the rebalance

[Pehat]

I have a consumer reading a single partition. The code is:

for message in self.consumer:
    result = make_love(message)  # up to 10 seconds; it's a long time, believe me!
    make_baby(result)
    try:
        self.consumer.commit()
    except IllegalGenerationError:
        logging.error('Illegal generation caught! Skipping it, huh')

As far as I know, it goes as follows:

1. Consumer picks the next message from the queue.
2. Heavy processing routine is executed.
3. GroupCoordintor performs rebalancing, thus incrementing the current generation number.
4. A new object is created using result as an argument.
5. Consumer commits the offset of the successfully processed message.
6. GroupCoordinator responses with IllegalGenerationError because while consumer has been processing the task, actual generation number has incremented so that the consumer becomes illegal.
7. Consumer reconnects, fetches the very message it has processed already (but failed to commit), resulting in spawning twin objects (undesired behaviour).

Meanwhile, the docs say the following:

The consumer connects to the co-ordinator and sends a HeartbeatRequest. If an IllegalGeneration error code is returned in the HeartbeatResponse, it indicates that the co-ordinator has initiated a rebalance. The consumer then stops fetching data, commits offsets and sends a JoinGroupRequest to it's co-ordinator broker. In the JoinGroupResponse, it receives the list of topic partitions that it should own and the new generation id for it's group. At this time, group management is done and the consumer starts fetching data and (optionally) committing offsets for the list of partitions it owns.

Of course, I can commit() at first and make_baby() thereafter, but in case of make_baby() failure it would be at-most-once processing, so it's an undesired behavior, too.

The question is: should the consumer be able to successfully commit its offsets if it gets an IllegalGenerationError? If not, how can I perform the exactly-once semantics?

[dpkp]

I suggest tuning your session timeout to be >> worst case message
processing time.

The docs refer to committing offsets during a rebalance. Your error comes
when you miss the rebalance entirely and commit after a new group is formed
without your consumer in it.
On Feb 10, 2016 6:48 AM, "Pehat" notifications@github.com wrote:

I have a consumer reading a single partition. The code is:

for message in self.consumer:
result = make_love(message) # up to 10 seconds; it's a long time, believe me!
make_baby(result)
try:
self.consumer.commit()
except IllegalGenerationError:
logging.error('Illegal generation caught! Skipping it, huh')

As far as I know, it goes as follows:

1. Consumer picks the next message from the queue.
2. Heavy processing routine is executed.
3. GroupCoordintor performs rebalancing, thus incrementing the current
   generation number.
4. A new object is created using result as an argument.
5. Consumer commits the offset of the successfully processed message.
6. GroupCoordinator responses with IllegalGenerationError because while
   consumer has been processing the task, actual generation number has
   incremented so that the consumer becomes illegal.
7. Consumer reconnects, fetches the very message it has processed already
   (but failed to commit), resulting in spawning twin objects (undesired
   behaviour).

Meanwhile, the docs
https://cwiki.apache.org/confluence/display/KAFKA/Kafka+0.9+Consumer+Rewrite+Design#Kafka0.9ConsumerRewriteDesign-Consumer
say the following:

The consumer connects to the co-ordinator and sends a HeartbeatRequest. If
an IllegalGeneration error code is returned in the HeartbeatResponse, it
indicates that the co-ordinator has initiated a rebalance. The consumer
then stops fetching data, commits offsets and sends a JoinGroupRequest
to it's co-ordinator broker. In the JoinGroupResponse, it receives the list
of topic partitions that it should own and the new generation id for it's
group. At this time, group management is done and the consumer starts
fetching data and (optionally) committing offsets for the list of
partitions it owns.

Of course, I can commit() at first and make_baby() thereafter, but in
case of make_baby() failure it would be at-most-once processing, so it's
an undesired behavior, too.

The question is: should the consumer be able to successfully commit its
offsets if it gets an IllegalGenerationError? If not, how can I perform
the exactly-once semantics?

—
Reply to this email directly or view it on GitHub
#544.

[Pehat]

In my situation there is one and only consumer. In this case, if I understand the docs right, there is no way to perform a rebalance:

The co-ordinator does not allow offset commit requests with generation ids older than the current group generation id in zookeeper either. This constraint is not a problem during a rebalance since until all consumers have sent a JoinGroupRequest, the co-ordinator does not increment the group's generation id. And from that point until the point the co-ordinator sends a JoinGroupResponse, it does not expect to receive any OffsetCommitRequests from any of the consumers in the group in the current generation. So the generation id on every OffsetCommitRequest sent by the consumer should always match the current generation id on the co-ordinator.

So that if the new group is formed, it contains zero consumers, right?

[dpkp]

Correct. As I understand it, the JoinRequest closes when all consumers have checked-in, or when the session timeout expires. This is done to handle client failures. So here the long delay you have when processing is likely extending beyond the session timeout, which makes the coordinator think the client has failed.

[dpkp]

Closing because this failure mode is expected.

[jack8daniels2]

Is increasing session_timeout_ms the recommended solution? Is there a way to let the consumer send heartbeats from a different thread, while the main thread is busy processing?

[dpkp]

Yes. For current discussion of alternative proposal, see:

https://cwiki.apache.org/confluence/display/KAFKA/KIP-62%3A+Allow+consumer+to+send+heartbeats+from+a+background+thread

Discussion is on the dev kafka mailing list if you are interested.