# Consumer Groups, Partitions and Offsets

`03-kafka/` introduced these three ideas; this topic is the full mechanics, because almost every Kafka production issue — lag, duplicates, stalled consumers, "rebalance storms" — comes down to how they interact.

## 1. Partitions Are the Unit of Parallelism

A topic is split into **partitions**; each partition is an ordered, append-only log. Within a **consumer group**, each partition is assigned to **exactly one** consumer at a time:

```
topic "orders" (6 partitions)            group "billing" (3 consumers)
  P0 P1 ─────────────────────────────►    consumer A
  P2 P3 ─────────────────────────────►    consumer B
  P4 P5 ─────────────────────────────►    consumer C
```

Consequences:

- **Max useful consumers per group = number of partitions.** A 7th consumer on a 6-partition topic sits idle.
- **Ordering is per partition**, and since one consumer owns a partition, that consumer sees its keys in order.
- **Different groups are independent**: group "billing" and group "analytics" each get every message, each with their own progress.

## 2. Offsets: Where Each Group Is

Each record in a partition has an **offset** (0, 1, 2, …). A group's progress is stored as the **committed offset** per partition in the internal `__consumer_offsets` topic — and it's the offset of the **next** record to read, not the last one processed.

```
partition P0:  [0][1][2][3][4][5][6][7][8]
                            ▲           ▲
              committed = 4 ┘           └ log end offset = 9
              lag = 9 − 4 = 5
```

**Consumer lag** (log end offset − committed offset, summed over partitions) is the single most important consumer health metric: steadily growing lag means consumers can't keep up.

## 3. Committing

| Strategy | How | Risk |
|---|---|---|
| auto-commit (`enable.auto.commit`) | commits the last *polled* position every N ms | may commit messages not yet processed (loss on crash) or reprocess after a crash (dupes) |
| commit after processing (sync) | `commitSync()` after a batch | at-least-once; slower |
| async commit | `commitAsync()` | fast; an older commit can land after a newer one — use sync on shutdown/rebalance |
| store offsets with the results | write output + offset in one DB transaction | exactly-once *effect* without Kafka transactions |

Where a *new* group starts is controlled by `auto.offset.reset`: `earliest` (replay everything retained) or `latest` (only new messages).

## 4. Rebalancing

When a consumer joins, leaves, crashes, or misses its heartbeat/poll deadline, the group **rebalances**: partitions are reassigned among the live members.

- **Eager rebalance** (classic): every consumer revokes *all* partitions, then gets a new assignment — the whole group pauses.
- **Cooperative/incremental rebalance** (`CooperativeStickyAssignor`): only partitions that actually move are revoked; everyone else keeps consuming.
- **Static membership** (`group.instance.id`): a consumer that restarts quickly (a rolling deploy) gets its old partitions back without triggering a rebalance.

Before giving up a partition, a consumer should **commit what it has processed** (the `onPartitionsRevoked` callback); otherwise the new owner starts from an older commit and reprocesses.

## 5. Choosing the Partition Count

- Enough for peak parallelism: if one consumer handles 2,000 msg/s and peak is 50,000 msg/s, you need ≥ 25 partitions (plus headroom).
- Increasing partitions later **changes key → partition mapping** (`hash(key) % N`), which breaks per-key ordering across the change — so over-provision moderately up front.
- Too many partitions cost broker memory, file handles, and leader-election time.

## Under the Hood: The Group Coordinator Protocol

One broker acts as the **group coordinator** for each group (chosen by hashing the group id onto a `__consumer_offsets` partition). Membership works like this:

1. Each consumer sends **JoinGroup**. The coordinator waits for members, then picks one as the **group leader**.
2. The leader (a *client*, not the broker) runs the configured **assignor** (range, round-robin, sticky, cooperative-sticky) over all members' subscriptions and sends the result in **SyncGroup**; the coordinator distributes each member's share.
3. Members send **heartbeats** (from a background thread). If no heartbeat arrives within `session.timeout.ms`, or a member doesn't call `poll()` within `max.poll.interval.ms`, it's evicted and a new rebalance starts — this is how a stuck processing loop turns into a "rebalance storm".
4. Each assignment carries a **generation id**. Commits from an older generation are rejected, so a zombie consumer that was evicted can't overwrite the new owner's progress.

Offset commits are just records written to `__consumer_offsets` keyed by `(group, topic, partition)`; that topic is log-compacted, so only the latest committed offset per key is retained.

![Six partitions are split across a group's three consumers while a second group reads the same partitions independently; each group commits its own offset per partition, and lag is the distance to the log end](diagrams/01-groups-and-offsets.svg)

## Common Pitfalls

1. **More consumers than partitions** — idle consumers, no extra throughput.
2. **Long processing between polls** — `max.poll.interval.ms` exceeded → eviction → rebalance → duplicates.
3. **Committing before processing** (or careless auto-commit) → message loss.
4. **Not committing on revoke** → reprocessing after every rebalance.
5. **Adding partitions to a keyed topic** without planning — per-key ordering breaks during the transition.
6. **Monitoring throughput but not lag** — lag is what tells you you're falling behind.

## Quick Reference

| Concept | Rule |
|---|---|
| partition ↔ consumer | one consumer per partition per group |
| parallelism cap | #partitions |
| committed offset | next offset to read |
| lag | log end − committed |
| new group starts at | `auto.offset.reset` = earliest / latest |
| safer rebalances | cooperative-sticky + static membership |
| zombie protection | generation id on commits |