Explain the __consumer_offsets topic: why it's log-compacted, how offsets are keyed, and how the group coordinator uses it.

## What it is `__consumer_offsets` is an **internal, auto-created** Kafka topic that durably persists consumer group state — primarily committed offsets, plus group metadata. It's a normal Kafka topic used as a key-value store via the log. ## Why a topic (not a separate database) Kafka reuses its own replicated, durable log instead of an external store. Commits become **append-only messages**; durability and replication come for free from Kafka's partition replication. ## Partitions and replication - Default **50 partitions** (`offsets.topic.num.partitions`). - Default replication factor **3** (`offsets.topic.replication.factor`), so group state survives broker loss. A consumer group maps to exactly one partition of this topic via `hash(groupId) % offsets.topic.num.partitions`. The **leader** of that partition is the group's **coordinator** — the broker responsible for that group's offset commits/fetches and rebalance orchestration. ## Keying Each committed offset is a message whose **key is (groupId, topic, partition)** and whose **value** encodes the committed offset (next-to-read), optional metadata, leader epoch, and commit timestamp. Because the key uniquely identifies the slot being updated, repeated commits to the same slot share a key. ## Why log compaction The topic is configured `cleanup.policy=compact`. **Log compaction** retains, for each distinct key, **only the latest message**, discarding older ones in the background. Without compaction, a busy group committing every few seconds would write millions of obsolete offset records forever. With compaction, the topic's size is bounded to roughly one record per active (group,topic,partition). A **tombstone** (null value) for a key deletes that key entirely — used to expire offsets for groups/partitions no longer in use, governed by `offsets.retention.minutes`. ## The coordinator's use of it 1. **OffsetCommit**: a consumer's commit becomes an append to the coordinator's __consumer_offsets partition; once replicated, the commit is durable. 2. **OffsetFetch**: on (re)join, the consumer asks the coordinator for its committed offsets; the coordinator serves them from its in-memory map. 3. **Recovery**: if a coordinator broker restarts or leadership moves, the new leader **replays** the __consumer_offsets partition from the log to rebuild the in-memory offset/group map before serving requests. ## Practical notes - You can inspect it with `kafka-console-consumer --topic __consumer_offsets --formatter kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter` (or the consumer-group tooling, which is friendlier). - `kafka-consumer-groups.sh --describe --group X` reads from here to show current offset, log-end offset, and lag. - Don't manually produce to it; it's managed by the coordinator.