Kafka at Scale: Tiered Storage, KRaft, and Exactly-Once

Kafka's central abstraction is the partition log: an immutable, append-only sequence of records ordered by offset. This is not an implementation detail — it is the design decision that explains everything else. Traditional queues (SQS, RabbitMQ) delete messages after consumption; Kafka retains them. This allows multiple consumer groups to read the same data independently, deterministic reprocessing, and native auditing. The price is that the broker must actively manage retention, and local disk becomes a bottleneck in clusters with high volume or long retention.

Each topic is divided into partitions, each partition replicated across brokers according to the configured replication factor. The leader broker for each partition accepts writes; followers replicate synchronously (if in the ISR — In-Sync Replica set). A producer writes to the leader; the record is only considered committed when all ISR members confirm — that is Kafka's durability contract. The acks=all (or acks=-1) parameter combined with min.insync.replicas defines the minimum confirmation quorum. Without understanding this triangle (leader, ISR, acks), any discussion of exactly-once is premature.

The segment model is where Kafka gains I/O efficiency: each partition on disk is a sequence of segments (.log files + offset and timestamp indexes). The broker only writes to the active segment (tail); closed segments are immutable. This enables sendfile() directly from the filesystem cache to the consumer's socket — zero-copy — without passing through the JVM heap. In high-throughput workloads, Kafka frequently saturates the network before the disk, precisely because of this design.

ZooKeeper was the pragmatic choice of 2011: a mature distributed coordination service that Kafka used to store cluster metadata (which brokers are alive, which are partition leaders, topic configurations, ACLs). It worked for a decade, but created an expensive architectural coupling: two distributed systems to operate, two security models to configure, and a real scaling bottleneck — ZooKeeper stored all metadata in memory, and clusters with hundreds of thousands of partitions began experiencing latency in controller election and metadata propagation.

KRaft (Kafka Raft Metadata) solves this via KIP-500: Kafka brokers themselves now run an internal Raft protocol for metadata consensus. A subset of brokers (or dedicated controller nodes) forms the controller quorum; the active controller maintains a replicated metadata log — the same log/segment model as Kafka, applied to itself. This eliminates ZooKeeper as an operational dependency and, more importantly, removes the practical partition-per-cluster limit: Confluent benchmarks showed controller failure recovery in seconds with KRaft vs. tens of seconds with ZooKeeper on large clusters.

The transition was not trivial. KIP-500 was proposed in 2019; KRaft entered early access in Kafka 2.8 (2021), reached GA in 3.3 (2022) for new clusters, and ZooKeeper-to-KRaft migration support arrived in 3.5. ZooKeeper was officially removed in Kafka 4.0 (2024). On Amazon MSK, KRaft clusters are available from Kafka 3.6 on MSK. The critical operational point: in KRaft mode, controller nodes need reliable disks and low network latency between them — they are the new control plane of the cluster. In MSK, this is abstracted; in self-managed clusters, it is an explicit infrastructure decision.

KIP-405 (Tiered Storage) solves a real economic problem: local disk on Kafka brokers is expensive and does not scale independently of compute. In clusters with days or weeks of retention, most data is cold — rarely read, but must be available for reprocessing or auditing. Keeping this on local EBS or NVMe is wasteful.

The mechanism works as follows: the RemoteLogManager (RLM), the component introduced by KIP-405, monitors closed segments for each partition. When a segment meets the offload criteria (configurable by time or size), the RLM uploads the .log file, offset and timestamp indexes, and metadata to remote storage — in MSK's case, Amazon S3. After upload confirmation, the local segment can be deleted, freeing disk. The broker maintains a remote metadata map indicating which offsets are in S3 vs. local.

From the consumer's perspective, the protocol is transparent: it issues a normal FetchRequest to the leader broker. If the requested offset is in remote storage, the broker fetches the data from S3, optionally caches it locally (configurable), and returns it to the consumer. The consumer does not know the data came from S3. This is elegant, but has latency implications: a cold data fetch has S3 latency (typically tens of ms) vs. local disk or page cache (sub-ms). For batch reprocessing workloads, this is acceptable; for consumers requiring low latency on historical data, it is not.

On Amazon MSK, Tiered Storage is enabled via cluster configuration and uses AWS-managed S3 — you do not configure the bucket directly, but can use S3 Lifecycle Policies to control long-term storage costs. Current limitation: Tiered Storage on MSK does not support all instance types, and log compaction has restrictions with tiered storage enabled — compacted segments are not eligible for offload in certain versions. This is critical for changelog topics (Kafka Streams, CDC).

Exactly-once in Kafka is implemented in two complementary layers, and confusing the two is a common mistake in interviews and in production.

Layer 1 — Idempotent Producer: enabled with enable.idempotence=true (default true since Kafka 3.0). The broker assigns the producer a ProducerID (PID) and tracks the sequence number per partition. If the producer retransmits a message (due to timeout or network failure), the broker detects the duplicate sequence number and discards it — no duplicates in the log. This guarantees exactly-once per partition, per producer session. Cost: practically zero — it is bookkeeping on the broker, with no additional distributed coordination.

Layer 2 — Transactions: enabled with transactional.id on the producer. The producer coordinates with a Transaction Coordinator (a special broker elected by hash of transactional.id) to begin, commit, or abort a transaction that can span multiple partitions and multiple topics. The mechanism uses an internal two-phase commit: the coordinator writes the transaction state to an internal topic (__transaction_state) before instructing brokers to mark messages as committed. Consumers with isolation.level=read_committed only see messages from committed transactions — messages from open or aborted transactions are filtered.

The real cost of transactions is non-trivial: additional round-trip latency to the Transaction Coordinator, write overhead on __transaction_state, and the risk of zombie transactions — producers that failed without aborting the transaction, leaving it open until transaction.timeout.ms. In high-throughput systems, very large transactions (many messages per transaction) or very frequent transactions (one transaction per message) significantly degrade throughput. The practical recommendation: batch your transactions by time window or record count, not per individual message.

A frequently overlooked point: exactly-once end-to-end (producer → broker → consumer → sink) requires the sink to also be idempotent or transactional. Kafka guarantees exactly-once in the log; if the consumer processes and writes to a database without idempotency, you still have at-least-once in the system as a whole. Kafka Streams implements end-to-end exactly-once using Kafka transactions to atomically coordinate read, process, and write — but this only works when both source and sink are Kafka.

The consumer group model is what allows Kafka to scale consumption horizontally: each partition is assigned to exactly one consumer within the group, and multiple groups can consume the same topic independently. The Group Coordinator (a broker elected by hash of group.id) manages the group lifecycle: join, assignment synchronization, heartbeats, and failure detection.

The classic problem is rebalance: whenever a consumer joins, leaves, or stops sending heartbeats within session.timeout.ms, the coordinator triggers a rebalance. In the original protocol (Eager Rebalance), all consumers in the group stop consuming, revoke all partitions, and wait for the new assignment. In large groups with many partitions, this can cause pauses of seconds to tens of seconds — a distributed stop-the-world.

Kafka 2.4 introduced the Cooperative Sticky Assignor (KIP-429/KIP-482), which implements incremental rebalancing: only the partitions that need to be moved are revoked; the rest continue to be consumed during the rebalance. This drastically reduces the impact, but requires all consumers in the group to use a cooperative assignor — a configuration detail that is frequently forgotten during migrations.

Another critical operational point: max.poll.interval.ms defines the maximum time between two poll() calls. If processing a batch takes longer than this interval (heavy processing, slow external calls), the consumer is considered dead by the coordinator and a rebalance is triggered — even if the consumer is alive and processing. This creates a rebalance loop in workloads with variable processing time. The solution is not to increase max.poll.interval.ms indefinitely (this delays detection of real failures), but to reduce batch size via max.poll.records to ensure processing fits within the configured interval. On Amazon MSK, these parameters are configurable per consumer — MSK does not manage them; they remain entirely in your application.