Apache Kafka Conceptual

The reasoning

A **topic** is a logical category that producers write to and consumers subscribe to. It holds no data itself. The data lives in its **partitions**: each one is an ordered, immutable sequence of records, stored as segment files on the broker that leads it, with every record numbered by an **offset**. Offset 42 means nothing without a partition — partition 3 and partition 5 of the same topic both have an offset 42.

Partitions exist because a single log caps everything at one machine. One log means one leader broker taking every write and, within a consumer group, one consumer reading everything. Split the topic into 12 partitions and the leaders spread across brokers, and up to 12 consumers in a group can read at once. **The partition count is the ceiling on consumer parallelism** in a classic consumer group, because a partition is read by at most one member of the group.

The partition is also the unit of **replication** — each has its own leader, followers and in-sync replica set — and of **ordering**. Records within a partition are read in the order they were appended; across partitions there is no ordering relationship at all. That is why the record key matters: it decides the partition, and so it decides which records are ordered relative to each other.

Partitions are not free. Each one is open segment and index files on every replica, replication traffic, cluster metadata and a leader to move when a broker fails. And the count only goes up: you can add partitions to a topic but never remove them.

The reasoning

If the record names a partition explicitly, that wins. Otherwise, **with a key**, the built-in logic computes `murmur2(keyBytes)` and takes it modulo the number of partitions. The hash is over the **serialised bytes**, not the value in your code — so the string "42" and the integer 42 are different keys, and two services that serialise the same order ID differently put its events on different partitions and lose ordering between them without any error.

**Without a key**, older clients spread records round-robin, one record per partition in turn, which produced many tiny batches. Kafka 2.4 (KIP-480) switched to a sticky partitioner that fills a batch for one partition before switching, and Kafka 3.3 (KIP-794) made the built-in logic switch after at least `batch.size` bytes and steer away from slow brokers. In 4.x `partitioner.class` defaults to null, meaning that built-in behaviour, and `partitioner.ignore.keys=true` makes the producer ignore keys for placement altogether.

The consequence that matters in production: the mapping depends on the partition count. Add partitions and `hash % N` changes for most keys, so new records for a key go somewhere other than its old ones. A consumer that is behind can then process a key's newer events before its older ones.

A custom partitioner is the tool for deliberate placement — isolating a large tenant, salting one known hot key. The catch is that every producer to the topic must use the same one, or the same key lands in two places.

The reasoning

Consuming advances the group's **committed offset**, which is stored separately in the internal `__consumer_offsets` topic. The data itself is untouched, so ten groups can read the same record and any of them can go back and read it again by resetting its offset. That is the difference from a traditional queue, where an acknowledgement removes the message for everyone.

With the default `cleanup.policy=delete`, retention is by time — `retention.ms`, whose broker default is `log.retention.hours=168`, seven days — and optionally by size per partition, which is unlimited by default. Whichever limit is hit first applies.

The unit of deletion is the **segment**. Each partition is a series of segment files that roll at `segment.bytes` (1 GiB) or `segment.ms` (seven days). The active segment is never deleted, and a closed segment goes only when its newest record is older than the retention limit. So on a low-volume topic a record can outlive its retention by up to a segment's lifetime — a seven-day topic can hold data for close to fourteen.

Retention is also where consumers get hurt. A group that falls behind by more than the retention window comes back to an offset that no longer exists, and then `auto.offset.reset` decides what happens: with the default, `latest`, it skips silently to the end.

The reasoning

A consumer sends fetch requests to the leaders of its partitions, asking for records from its current position. The broker can hold the request until at least `fetch.min.bytes` is available (default 1 byte) or `fetch.max.wait.ms` passes (default 500 ms), so an idle consumer is not spinning. Each `poll()` then hands the application at most `max.poll.records` (500) from what was fetched.

Pulling was a deliberate choice. A push broker has to decide how fast to send to every consumer; a slow consumer is either overwhelmed or the broker has to track flow control for each one. With pull, a slow consumer simply falls behind: the backlog is visible as lag and bounded by retention, not by anyone's memory. Pull also lets a consumer fetch as much as is available in one request, which is what makes bulk reads cheap.

Because the consumer chooses the offset, **replay is free**: a group can seek back to yesterday, and a brand-new group can start at the beginning of the topic. None of that is natural in a broker that pushes and forgets.

The flip side is that Kafka does not slow producers down for slow consumers. Producers keep appending and the consumer's problem is its own — which is why lag, not backpressure, is the number you watch on a Kafka pipeline.

The reasoning

Every partition has a leader, and its followers fetch from it much like consumers do. The leader tracks how far each follower has got. A follower that has not fetched, or has not caught up to the leader's log end offset, for `replica.lag.time.max.ms` (30000 ms in 4.x) is dropped from the ISR, and it rejoins as soon as it catches up. The ISR is recorded in the cluster metadata the controller manages.

The **high watermark** is the offset up to which every current ISR member has replicated. Records past it exist on the leader but are not yet committed, and consumers are not given them. An acks=all write is acknowledged once the whole ISR has it — which is the same moment it falls under the high watermark.

A follower that stays out for hours cannot keep up: its broker is overloaded or unhealthy, its disk or network is saturated, it is stuck in garbage collection, or it is simply down. The consequence is that you have fewer real copies than the replication factor suggests. With replication factor 3 and an ISR of one, the next failure either takes the partition offline or — if unclean election is on — loses data. With `min.insync.replicas=2`, acks=all writes start failing at an ISR of one, which is the intended alarm.

The things to look at are the `UnderReplicatedPartitions` broker metric, ISR shrink and expand rates, `kafka-topics.sh --describe --under-replicated-partitions`, and the lagging broker's disk, network and garbage collection.

The reasoning

`acks=all` (the same as -1, and the producer default since Kafka 3.0, KIP-679) means the leader responds only when **every member of the current ISR** has the record. With all three replicas in sync that is three; if one follower has dropped out, it is two — and nothing tells the producer the number changed.

`min.insync.replicas` is a broker or topic setting that turns that into a guarantee. If the ISR is smaller than it when an acks=all write arrives, the leader rejects the write with `NOT_ENOUGH_REPLICAS`. It has no effect on acks=0 or acks=1. Replication factor 3 with a minimum of 2 tolerates one broker down for writes, and two down for reads.

With two brokers down, the survivor leads the partition (if it was in sync) and the ISR is just itself. Every acks=all write fails; the producer retries — `retries` defaults to effectively unlimited — until `delivery.timeout.ms` (120000 ms) expires, then the send callback gets the exception and the application has to decide what to do. Consumers keep reading everything up to the high watermark.

The broker default for `min.insync.replicas` is **1**, which means that out of the box an acks=all write can be acknowledged by the leader alone whenever the ISR has shrunk to it. Setting it to 3 on a 3-replica topic goes the other way: any single broker restart, including a routine rolling upgrade, stops all writes.

The reasoning

A normal election picks the new leader from the ISR, so the new leader has every committed record. The unclean case is when the ISR is empty — every in-sync replica has failed — and the only live replicas are followers that had fallen behind.

With the default, **false**, the partition simply has no leader. Producers and consumers get errors for that partition until an ISR member comes back. Consistency is kept; availability is lost for as long as the outage lasts.

With **true**, the controller elects one of the out-of-sync replicas. Every record it lacks — including ones acknowledged under acks=all — is lost, and when the old leader returns it truncates its log to match the new leader, so the divergence is permanent. The setting can be set per topic and changed dynamically, and `kafka-leader-election.sh --election-type unclean` forces a one-off unclean election for a specific partition when an operator decides the data loss is acceptable.

The decision is per topic. A metrics or logging topic may prefer availability; an orders or payments topic should keep the default and pair it with `min.insync.replicas=2`, so that the ISR rarely gets small enough for this to matter.

The reasoning

Since Kafka 4.0 the only mode is KRaft. Brokers send regular heartbeats to the active controller; when they stop for longer than the broker session timeout, the controller fences the broker. A **controlled shutdown** is gentler: the broker asks the controller to move its leaderships away first, which is why planned restarts cause far less disruption than crashes.

For each partition the dead broker led, the controller chooses a new leader from the ISR, shrinks the ISR, and writes those changes to the `__cluster_metadata` log, which every broker replicates and applies. Under ZooKeeper, the controller was a broker that had to push these changes out by RPC and reload state from ZooKeeper on failover — slow for large partition counts, and one of the reasons KRaft exists.

Producers sending to the old leader see errors such as `NOT_LEADER_OR_FOLLOWER` or connection failures, refresh metadata and retry within `delivery.timeout.ms`; with idempotence on (the default) a retried batch that had in fact been written is not written twice. Consumers see a pause, not a gap: the new leader was in the ISR, so it has everything up to the high watermark. If that broker also led partitions of `__consumer_offsets`, the group coordinators for those groups move too and their members rejoin.

When the broker returns it comes back as a follower, catches up and rejoins the ISR. Leadership moves back to the preferred replica through automatic leader rebalancing (`auto.leader.rebalance.enable`, on by default), so load evens out again.

The reasoning

Under ZooKeeper, broker registrations, topic configuration, partition assignments and ISR state lived in ZooKeeper, and one broker was elected controller through it. That meant two distributed systems to secure, size and operate, metadata changes pushed to brokers by RPC, and a controller failover that had to reload the whole cluster state from ZooKeeper — which got slow as partition counts grew.

In **KRaft**, dedicated controller nodes (or combined broker-controller nodes for small clusters) form a Raft quorum, usually three or five. One is the active controller. Metadata is an event log, `__cluster_metadata`, which brokers replicate and apply, so standby controllers already hold the state and failover is quick. KRaft was declared production-ready in 3.3.

The upgrade notes for 4.0 are explicit: ZooKeeper mode has been removed, and clusters in ZooKeeper mode must be migrated to KRaft before they can be upgraded. The migration runs on 3.x — 3.9 is the last 3.x line and the usual bridge — with the controller quorum provisioned alongside ZooKeeper, a dual-write phase, then finalisation, after which there is no way back. Kafka 4.0 also needs Java 17 for brokers, Connect and tools, and Java 11 for clients and Streams.

Anything that read ZooKeeper directly — old scripts, monitoring that inspected znodes, tooling still passing `--zookeeper` — stops working. Everything talks to the cluster through `--bootstrap-server` and the Admin API.

The reasoning

The failure it fixes: the producer sends a batch, the leader writes it, and the acknowledgement is lost to a timeout or a leader change. The producer retries — `retries` is effectively unlimited, bounded by `delivery.timeout.ms` — and without idempotence the batch is appended a second time. With it, the producer gets a **producer ID and epoch** at start-up and stamps each batch with a sequence number per partition; the broker remembers recent sequence numbers per producer and partition, acknowledges a duplicate without appending it, and rejects a gap.

It needs `acks=all`, retries above zero and `max.in.flight.requests.per.connection` of at most 5. It has been the default since 3.0 (KIP-679) — in practice from 3.0.1, 3.1.1 and 3.2.0, because a bug (KAFKA-13598) left it off in 3.0.0 and 3.1.0 — **as long as no conflicting setting is present**: set `acks=1` without explicitly enabling idempotence and it is quietly turned off. Explicitly setting `enable.idempotence=true` together with a conflicting setting (acks other than all, or more than 5 requests in flight) fails with a ConfigException rather than silently turning idempotence off.

Its scope is one producer session and one partition. A producer that crashes and restarts gets a new ID, so if the application re-sends records it was unsure about, those are new records to the broker. A caller that retries its own HTTP request and calls `send()` twice produces two records. A consumer that reprocesses after a rebalance has nothing to do with the producer at all.

So the idempotent producer gives you **exactly-once writes of a batch into a partition** — valuable, and much narrower than "no duplicates". For restarts, use a `transactional.id`, which gives the producer a stable identity across sessions; for everything else, deduplicate by key where the data lands.

The reasoning

The producer does not wait for batch 1 to be acknowledged before sending batches 2 to 5. Suppose the leader rejects batch 1 with a transient error but accepts batch 2. The producer retries batch 1, and it is appended after batch 2. The partition preserves the order records were **appended**, which is no longer the order they were **sent**. The Kafka producer documentation describes exactly this for `enable.idempotence=false` with more than one request in flight.

The old fix was `max.in.flight.requests.per.connection=1`: one request per round trip per connection, which is safe and throttles throughput badly on any real network.

The idempotent producer fixes it properly. Each batch carries a sequence number for its partition, and the broker appends only the next expected sequence. A later batch that arrives while an earlier one is being retried is rejected as out of sequence, and the producer re-sends in order. The broker keeps state for the last five batches per producer and partition, which is where the limit of five comes from.

Ordering can still break outside the producer: records for the same entity sent with different keys, two producer instances writing the same key with no coordination between them, or application code that catches a failed send and re-sends it itself instead of letting the client retry.

The reasoning

`initTransactions()` finds the coordinator — the leader of the `__transaction_state` partition that the transactional.id hashes to — which assigns a producer ID and epoch and aborts anything the previous holder of that ID left open. Then `beginTransaction()`, the sends (each new partition is registered with the coordinator), `sendOffsetsToTransaction(offsets, consumer.groupMetadata())` for the input positions, and `commitTransaction()`. The coordinator logs the decision, writes a commit marker into every partition in the transaction, and marks it complete.

A **zombie** is an instance that was paused — a long garbage collection, a network partition — and replaced, and then wakes up and tries to commit. The replacement's `initTransactions()` already bumped the epoch, so the broker rejects the old epoch with `ProducerFencedException`. Since KIP-447 fencing also uses the consumer group metadata, which is why `exactly_once_v2` in Kafka Streams needs brokers on 2.5 or later.

On the consuming side, `isolation.level` defaults to **read_uncommitted**, which returns everything including records from aborted transactions. With `read_committed`, a fetch returns records only up to the **last stable offset** — the first offset of any still-open transaction — and filters out aborted ones. A transaction left open therefore stalls read_committed consumers, which looks like lag even though the data is on disk; `transaction.timeout.ms` (60 seconds by default) bounds how long that can last.

The atomicity covers Kafka partitions and consumer offsets, nothing else. A database write or an API call inside the transaction is not rolled back when it aborts.

The reasoning

The rule in a consumer group is that a partition is assigned to **at most one member**, while one member may own several partitions. So the partition count caps the group's parallelism. With 2 partitions and 3 consumers, one consumer gets nothing: it still heartbeats, and it takes a partition over if another member leaves.

The other way round — 6 partitions, 2 consumers — each consumer owns three. Which three is decided by the assignment strategy. The client default in 4.x is the list `[RangeAssignor, CooperativeStickyAssignor]`, which resolves to range: contiguous ranges per topic, which can pile extra partitions onto the first consumers when a group subscribes to several topics.

A second group on the same topic receives every record again, tracked by its own committed offsets. This is how Kafka does fan-out: **queue semantics within a group, publish-subscribe across groups**. The groups never coordinate, and each has its own lag.

The exception is new. Share groups (KIP-932, production-ready in Kafka 4.2) let several consumers take records from the same partition, with per-record acknowledgement and delivery counts. They trade away per-partition ordering to get there, so they suit independent work items rather than ordered event streams.

The reasoning

Triggers: a member joins (scale-out, a restart), a member leaves or is presumed dead after `session.timeout.ms`, a member fails to call `poll()` within `max.poll.interval.ms`, the subscribed topics gain partitions, or a regex subscription matches a new topic. Static membership (`group.instance.id`) lets a restarted member reclaim its partitions without a rebalance if it returns within the session timeout.

The classic **eager** protocol is stop-the-world: every member revokes all of its partitions, rejoins, the group leader computes a new assignment, and everyone resumes. One pod restarting pauses the whole group, and the pause lasts as long as the slowest member takes to rejoin.

**Cooperative sticky** (KIP-429, Kafka 2.4) keeps every partition that is not moving with its owner; only moved partitions are revoked, over two quick rounds. The 4.x client default `[RangeAssignor, CooperativeStickyAssignor]` still resolves to range, so a group behaves eagerly until it is migrated — which takes two rolling bounces, first adding the cooperative assignor, then removing range.

**KIP-848** (`group.protocol=consumer`, GA in 4.0 and enabled on the broker once the upgrade is finalised) replaces the join-and-sync dance: the group coordinator on the broker computes assignments with a server-side assignor (`group.consumer.assignors`, default `uniform,range`) and hands out changes through heartbeats, with no group-wide barrier. The client still defaults to `classic`, and under the new protocol the session timeout and heartbeat interval are broker settings (45 s and 5 s by default).

The reasoning

Since Kafka 0.10.1 (KIP-62) heartbeats are sent from a **background thread**, independently of the application's poll loop. So a slow batch no longer misses heartbeats — and a hung process that still has a live heartbeat thread would never be detected by the session timeout alone. Hence two timeouts: `session.timeout.ms` (45000 ms since 3.0, with `heartbeat.interval.ms` at 3000 ms under the classic protocol) for "is the process alive", and `max.poll.interval.ms` (300000 ms) for "is it making progress".

When the gap between two `poll()` calls exceeds `max.poll.interval.ms`, the client leaves the group and a rebalance hands its partitions to someone else. When the slow batch finally finishes, the commit fails with `CommitFailedException`, so the new owner re-reads everything since the last successful commit — which is where the duplicates come from. For a static member (`group.instance.id`), the documentation says partitions are not reassigned at once; the consumer stops heartbeating and they move when the session timeout expires.

The tuning follows from the arithmetic: worst-case batch time is roughly `max.poll.records` (500 by default) times the per-record cost. Lower `max.poll.records`, raise `max.poll.interval.ms` to cover the real worst case, or hand the work to a pool while calling `pause()` on the partitions and keeping `poll()` going so the member stays in the group.

Under the KIP-848 protocol the session timeout moves to the broker (`group.consumer.session.timeout.ms`, 45 s) and the client's session and heartbeat settings are not supported, but `max.poll.interval.ms` still applies on the client.

The reasoning

Before Kafka 0.9 offsets lived in ZooKeeper. Now they are records in `__consumer_offsets`, a compacted topic with 50 partitions by default. A group's offsets live in the partition that its `group.id` hashes to, and the broker leading that partition acts as the **group coordinator** for the group — membership, rebalances and commits all go through it.

The number committed is the **next offset to consume**: last processed plus one. `commitSync()` with no arguments commits the positions returned by the last `poll()`, which already follow that rule; a manual commit of `record.offset()` without the +1 replays one record per partition on every restart.

Expiry: with `offsets.retention.minutes=10080` (seven days), the offsets of a group that has had no members for that long are deleted. When it comes back there is no committed offset, so `auto.offset.reset` decides: `latest` (the default) silently skips everything written since, `earliest` re-reads the whole retention window, `none` throws. Recent versions also accept `by_duration:<duration>` to start a fixed period back.

Keep two numbers apart. The consumer's **position** moves every time it polls; the **committed** offset moves only when it commits. Lag is the log-end offset minus the committed offset, per partition. And offsets do not have to live in Kafka at all — a consumer can store them with its output and `seek()` to them when partitions are assigned.

The reasoning

With `enable.auto.commit=true` (the default) and `auto.commit.interval.ms=5000`, the commit happens **inside `poll()`** (and `close()`), and it commits the offsets of records returned by the **previous** poll. In a plain single-threaded loop — poll, process everything, poll again — that is at-least-once: a crash replays up to a few seconds of work. It turns into loss the moment you hand records to other threads and poll again before they finish, because the next poll commits offsets for work still in flight.

Commit manually (`enable.auto.commit=false`) when processing is asynchronous or multi-threaded, when you batch writes and should commit only after the flush, when the commit has to line up with a write to a sink, or when you need per-partition control.

`commitSync()` blocks until the commit succeeds, retrying retriable errors, and throws on fatal ones such as `CommitFailedException` after the group has moved on. `commitAsync()` returns at once with an optional callback and does **not retry**, because a retried older commit could land after a newer one and move the offset backwards. The common pattern: `commitAsync()` on each loop, `commitSync()` in the shutdown path and inside `ConsumerRebalanceListener.onPartitionsRevoked`, so the next owner of a partition starts in the right place.

Commit per batch rather than per record: each commit is a write to `__consumer_offsets`, and committing every record multiplies that load for no gain. When committing explicit offsets, commit `offset + 1`.

The reasoning

Apache Kafka brokers store keys and values as bytes and never look inside them. **Schema Registry** (Confluent's, or a compatible implementation) is a separate HTTP service that stores versioned schemas under **subjects** — by default `<topic>-value` and `<topic>-key`.

On the wire, a record written by the Confluent Avro serialiser is a magic byte (0), a 4-byte schema ID, then the Avro binary. The schema itself is not in the message. The consumer's deserialiser looks the ID up once, caches it, and decodes using the writer's schema, resolved against its own reader schema.

Compatibility is enforced when the producer registers a new version. Confluent's documentation gives **BACKWARD** as the default — consumers using the new schema can read data written with the old one, so consumers upgrade first — because it lets consumers rewind to the start of a topic. FORWARD means old consumers can read new data, so producers upgrade first; FULL means both; the `_TRANSITIVE` variants check against every earlier version instead of only the latest; NONE turns the check off. Under BACKWARD in Avro you may add a field with a default or delete a field; adding a required field is rejected.

The check lives in the client and the registry, so it is only as strong as who can register. The serialiser defaults to `auto.register.schemas=true`, which lets any producer register any compatible schema; production teams often turn that off and register through CI instead.

The reasoning

In **distributed mode** several worker processes with the same `group.id` form a group that shares out connectors and tasks and rebalances when a worker joins or dies. Connectors are created through the REST API, and the cluster keeps its state in three compacted topics named by `config.storage.topic`, `offset.storage.topic` and `status.storage.topic`. **Standalone mode** is one process with offsets in a local file — fine for development, with no fault tolerance.

A **connector** decides how to split the work and hands out task configurations up to `tasks.max`. A JDBC source reading ten tables can run up to ten tasks. A sink task is a consumer in a group named after the connector, so sink parallelism is capped by the partitions it reads.

**Converters** (`key.converter`, `value.converter` — Avro, JSON, Protobuf) turn Connect's internal records into bytes and back, and are where Schema Registry plugs in. **Single message transforms** make light per-record changes; Debezium users commonly unwrap the change envelope to the row's new state with one.

Debezium's Postgres connector reads the write-ahead log through one logical replication slot, which is one ordered stream, so its documentation says it always uses a single task. Its source offset is a log position (LSN) stored in the offsets topic. The operational risk is not speed but the slot: a stalled connector makes Postgres retain WAL until the database disk fills. Throughput scales by splitting tables across connectors, each with its own slot.

The reasoning

A **Kafka Streams** application has no cluster of its own. You deploy your JVM service N times; the topology is split into tasks, one per input partition, and the instances share tasks through the consumer-group protocol. Stateful operations — aggregations, joins, windows — use **state stores**, local RocksDB by default, each backed by a compacted **changelog topic**. When a task moves, its new owner rebuilds the store by replaying the changelog. `num.standby.replicas` defaults to 0; setting it to 1 keeps a warm copy elsewhere so failover does not start from nothing.

The processing guarantee is a setting: `processing.guarantee` defaults to `at_least_once`, and `exactly_once_v2` turns the read-process-write loop into Kafka transactions. Re-keying writes to an internal repartition topic so the new key is co-located.

A **plain consumer** gives you none of that: state, recovery and exactly-once are yours to build. **Spark Structured Streaming** runs on a driver and executors, reads Kafka with `readStream.format("kafka")`, and records the offset range of every micro-batch in its **checkpoint**; it does not rely on the group's committed offsets. Its state store also lives under the checkpoint. Restart from the checkpoint and it resumes exactly; delete it and it starts again from `startingOffsets`.

The choice follows. Kafka Streams fits Kafka-in, Kafka-out services with per-key state and no appetite for another cluster. Spark fits jobs that join streams with lake tables and write to Delta or Iceberg. Flink fits very large state and event-time-heavy work. Kafka Streams also reads from and writes to Kafka only; anything else goes through Connect.

The reasoning

A Kafka **cluster** is a set of **brokers**, servers that store data and serve clients. Data is organised into **topics**, each split into **partitions**: ordered, append-only logs stored as segment files. Every partition is replicated to several brokers; one replica is the **leader**, which handles reads and writes, and the rest are followers that copy it. The followers that are caught up form the in-sync replica set, which decides what counts as committed.

**Producers** send records to topics, choosing the partition from the key (hash of the key) or spreading keyless records across partitions, and decide their durability with `acks`. **Consumers** pull records from partitions starting at an offset; in a **consumer group** each partition goes to one member, and the group's progress is stored as committed offsets in the internal `__consumer_offsets` topic.

The **controller** manages the cluster: broker membership, partition leadership, topic metadata. Since Kafka 4.0 this is always a KRaft quorum of controller nodes keeping metadata in an internal log; older clusters used ZooKeeper for it.

Around the core sit Kafka Connect (moving data in and out with configured connectors), Kafka Streams (a library for processing topics into topics) and, from Confluent rather than Apache Kafka, Schema Registry.

The reasoning

A **consumer** subscribes to topics (or is assigned partitions directly), calls `poll()` to fetch records from its current position in each partition, and commits offsets to record its progress.

A **consumer group** is every consumer with the same `group.id`. A broker acting as group coordinator tracks membership through heartbeats and hands out partitions so each goes to exactly one member. The group commits its offsets as a unit of progress: if a member dies, its partitions move to others, who resume from the committed offsets.

That gives Kafka two behaviours from one mechanism: **within** a group, records are shared out like a queue; **across** groups, every group gets every record, like publish-subscribe. It also caps a group's parallelism at the partition count — extra members sit idle — with share groups, production-ready in 4.2, as the exception for work that does not need per-partition order.

The reasoning

The **replication factor** is fixed per partition when the topic is created (or changed by a reassignment). With a factor of 3, three brokers hold a replica.

The **ISR** is dynamic. A follower stays in it while it keeps up with the leader — fetching up to the leader's log end within `replica.lag.time.max.ms`, 30 seconds by default — and drops out when it cannot. The ISR decides the **high watermark** (what consumers may read), which replicas an `acks=all` write waits for, and which replicas may become leader without losing data.

`min.insync.replicas` turns the ISR into a durability floor: if the ISR is smaller, acks=all writes are refused. Watching the under-replicated partition count is how you see the gap between the replication factor you configured and the redundancy you actually have.

The reasoning

A background **log cleaner** rewrites closed segments, keeping only the newest record for each key. The **active segment is never compacted**, and a partition is cleaned only once enough of it is superseded (`min.cleanable.dirty.ratio`, 0.5 by default). `min.compaction.lag.ms` (0 by default) guarantees a window in which every update stays visible; `max.compaction.lag.ms` forces cleaning within a bound.

A **tombstone** — a key with a null value — marks the key deleted. It is itself removed after `delete.retention.ms` (one day by default), so a consumer that reads a compacted topic more slowly than that can miss the delete entirely.

Offsets survive compaction, leaving gaps, and order per key is preserved. Compaction is **eventual**, not a deduplication guarantee: a consumer can still read several values for one key in the part not yet cleaned. A compacted topic rejects records without a key. And `cleanup.policy=compact,delete` combines both, dropping old segments by time or size as well — which bounds disk but means the topic no longer holds complete state.

The reasoning

In ZooKeeper mode, brokers registered themselves in ZooKeeper, topic and partition configuration and ISR state were stored there, and brokers competed to become the single **controller** through it. The controller — a broker, not ZooKeeper — chose partition leaders and pushed changes out. Consumer offsets had already moved out of ZooKeeper into `__consumer_offsets` in Kafka 0.9.

In **KRaft** mode the metadata is a Kafka log, `__cluster_metadata`, replicated by a quorum of controller nodes using Raft. Standby controllers already hold the state, so failover does not need a full reload, and there is one system to run instead of two. KRaft was declared production-ready in 3.3; Kafka 4.0 supports nothing else, and ZooKeeper clusters must be migrated before upgrading.