What will happen if librdkafka can't commit the offsets to the broker?

[JSoet]

From what I understand, when enable.auto.commit is true, then librdkafka will commit the latest offset of the messages which have been consumed by the application, and it'll reach out to the broker every auto.commit.interval.ms to do that.

But what happens if it can't reach the broker? I assume it will keep retrying? Would there be any error messages in the logs at this point if it wasn't able to reach the broker?

And at what point will the consumer be disconnected from the group? I assume if the consumer can't commit the offset, then it is also likely not able to poll, so would it just keep retrying to commit until max.poll.interval.ms is reached and then the consumer is disconnected? Or could the commit failing cause the librdkafka consumer to consider itself connected on its own and "disconnect" and "rejoin" the group?

And then when it reconnected it would then start from the last committed offset, so those messages which couldn't be commited could end up being duplicated, is that correct?

[edenhill]

Yes, it will retry offset commits, and if it fails to commit offsets it will trigger an optional offset_commit_cb() which propagates results of the automatic commits, and if that's not available a line will be logged, something like:
Offset commit (<commit-reason>) failed for x/y partitions in join-state <some-cgrp-state>: <some error message>: <partitions..>

The consumer will be (temporarily) evicted from the group on rebalance and if max.poll.interval.ms is triggered, or there's a connectivity issue with the group coordinator. The consumer will rejoin as soon as it can.

A consumer will be able to commit its final offsets during a rebalance, but not when being kicked out of the group for the other reasons.

When the consumer has rejoined the group it will read the last committed offsets for the partitions assigned to it.

[JSoet]

Thanks for the reply...

The error message that you mentioned would be an error level log message, correct?

And just to understand better, what do you mean by and here: "The consumer will be (temporarily) evicted from the group on rebalance and if max.poll.interval.ms is triggered", do you mean if the max.poll.interval.ms is hit, then a rebalance will be triggered? Or those are 2 separate cases where the consumer can be evicted from the group?

And finally, I have just a few follow-up questions about what happens in different cases when the broker goes down:

First, sorry, I don't really understand the difference between what would happen if a single broker went down vs if the consumer can't connect to any broker? And also, I'm wondering about the case where a single broker was shutdown gracefully vs if it went down unexpectedly.... Would the same log message you mentioned above still be printed if the consumer was rebalanced to another broker?

To expand on that, if a broker is shutdown gracefully, then I assume that triggers a rebalance and you said then there is a final chance to commit the offsets - if we're just using the enable.auto.commit=true and enable.auto.offset.store=true then I assume those commits could still be committed before it shutdown, and the broker would take care of making sure they get synchronized to the other brokers before shutting down? So in normal operations there shouldn't be any risk of duplicate messages happening when a broker is shutdown? Is that correct?

But what about if a single broker was unexpectedly stopped? I assume that would trigger a rebalance, correct? After the rebalance, could those uncommitted message offsets then be duplicated, because it would start at the offsets which were committed to that broker? Or is there anyway the consumer could know that it already consumed those messages, even though it's now connected to a different broker? I assume that in that case the messages could be duplicated, correct? Is a log message printed when the rebalance happens, and if so what log level is it?

(To explain the context - we had a situation where we ran into some duplicate messages in our application, and we're trying to piece together what might've happened. Unfortunately not all logging was enabled, only the warning and error level messages were being printed, and the kafka broker logs were already overwritten. I wrote up a separate discussion thread for that, just in case anyone had any other ideas around it)

[edenhill]

The error message that you mentioned would be an error level log message, correct?

I'll be a LOG_WARNING.

And just to understand better, what do you mean by and here: "The consumer will be (temporarily) evicted from the group on rebalance and if max.poll.interval.ms is triggered",

that should be "or if max.poll.. "

Brokers going down typically do not trigger a rebalance, the group coordinator duty will be automatically transitioned to another broker and the consumers will use FindCoordinatorRequest to find the new coordinator and continue their heartbeats.

There's no log message for rebalances, it is considered a natural and common part of the consumer group lifespan.

Duplicate messages may happen if a consumer is unable to commit offsets prior to loosing its assigned partitions (due to rebalance, disconnect, timeout, etc).

[JSoet]

Thanks once again, sorry for the continuing questions, I'm still just trying to make sure I understand here...

Brokers going down typically do not trigger a rebalance, the group coordinator duty will be automatically transitioned to another broker and the consumers will use FindCoordinatorRequest to find the new coordinator and continue their heartbeats.

So in that case even when the consumer switched to connect to a different broker then if you're using the auto.commit then it will b e able to commit those offsets to the other broker? So that should not result in any duplicates?

Duplicate messages may happen if a consumer is unable to commit offsets prior to loosing its assigned partitions (due to rebalance, disconnect, timeout, etc).

But by default does it commit any uncommitted offsets when a rebalance happens (even if there's no rebalance cb created and we're just using the auto.commit) and duplicates on a rebalance would just be an exceptional case if it was unable to commit the offsets?

And in all those cases (rebalance, disconnect, timeout, etc) then if it was unable to commit it would print the warning message you posted in your original reply?

[JSoet]

@edenhill Sorry to bother you, just was still wondering about the above if you have time