How does a ConsumerRebalanceListener's callback behavior differ between eager and cooperative rebalancing, and where should you commit offsets?

## The listener `ConsumerRebalanceListener` is an interface you pass to `consumer.subscribe(topics, listener)`. It has three callbacks the consumer invokes during rebalances: - **`onPartitionsRevoked(partitions)`** — called *before* the consumer loses ownership of `partitions`. This is the canonical place to **commit offsets** (and flush any local state / external buffers) so whoever picks up the partition continues from the right point and you avoid reprocessing. - **`onPartitionsAssigned(partitions)`** — called *after* the consumer gains `partitions`, before the next `poll()` returns records for them. Use it to initialize per-partition state, restore caches, or `seek()` to a custom offset. - **`onPartitionsLost(partitions)`** — called when partitions were lost **without** a graceful revoke — e.g. the member was fenced out (a newer generation took over) or it exceeded `max.poll.interval.ms`. The default implementation calls `onPartitionsRevoked`, but you typically override it to **NOT commit**, since you no longer safely own those partitions and committing could clobber the new owner's progress. ## How the protocol changes the callbacks ### Eager Every rebalance, before anything else, the consumer revokes **everything**. So `onPartitionsRevoked` receives the consumer's **entire** current assignment each time, and `onPartitionsAssigned` later receives the **entire** new assignment — even partitions it already had and kept. Callbacks are 'big' and fire on every rebalance for all partitions. ### Cooperative The consumer keeps the partitions it retains. So `onPartitionsRevoked` receives **only the partitions actually being moved away** (often empty), and `onPartitionsAssigned` receives **only the newly added** partitions. Retained partitions don't appear in either callback and never pause. This is what makes cooperative rebalances cheap — and it means your revoke logic should be written to handle a *subset*, not assume it's the whole assignment. ## Where to commit - **Auto-commit (`enable.auto.commit=true`):** the consumer auto-commits before `onPartitionsRevoked` for you. Fine for at-least-once with simple processing. - **Manual commit:** call `consumer.commitSync(offsets)` inside `onPartitionsRevoked`. Use `commitSync` (not async) here because you're about to lose the partitions and want the commit to complete first. - **Never commit in `onPartitionsLost`** for the lost partitions — ownership is already gone. ## Edge cases - In cooperative mode, since revoke is incremental, a long-running, blocking commit in `onPartitionsRevoked` still blocks the poll loop — keep it tight. - `onPartitionsAssigned` is also where you should re-seek if you store offsets externally (e.g. in a DB) rather than in Kafka. - If you do external transactional processing, the revoke callback is your last chance to flush atomically.