01. Distributed Message Queue — High-Level Design¶

~15 min read · Part 2 of 4 (Overview → HLD → LLD → Q&A)

This file turns the log abstraction from the overview into brokers, partitions, and the flows between producers and consumers. Read the architecture top to bottom, follow a publish and a consume through it, then watch what happens when a broker dies mid-stream and when a consumer falls a day behind.

Architecture¶

   producers                                             consumer groups
  ┌─────────┐                                          ┌──────────────────┐
  │ payments│──┐                                     ┌▶│ fraud-detector   │  (reads full stream)
  └─────────┘  │                                     │ └──────────────────┘
  ┌─────────┐  │   key→partition                     │ ┌──────────────────┐
  │ ledger  │──┼──▶ hash(key) % N                     ├▶│ warehouse-loader │  (day behind, catching up)
  └─────────┘  │        │                             │ └──────────────────┘
               ▼        ▼                             │ ┌──────────────────┐
        ┌───────────────────────────────────────┐    ├▶│ dashboard        │
        │            BROKER CLUSTER (~20)         │   │ └──────────────────┘
        │                                         │   │
        │  topic "txns", 200 partitions, RF=3     │   │
        │                                         │   │
        │  P0  leader:brokerA  [====offsets===▶]  │───┘   each consumer owns a
        │      followers: B, C  (ISR)             │        disjoint set of partitions
        │  P1  leader:brokerB  [====offsets===▶]  │        within its group; commits
        │      followers: C, A  (ISR)             │        its own offsets
        │  ...                                    │
        │  P199 leader:brokerT [===offsets===▶]   │
        └───────────────────┬─────────────────────┘
                            │ metadata: leaders, ISR, configs
                            ▼
                 ┌────────────────────────┐
                 │  Metadata quorum         │  (KRaft / ZooKeeper):
                 │  controller + raft log   │  leader election, membership
                 └────────────────────────┘

Read it top to bottom. Producers pick a partition by hashing the event key, so all events for one account_id flow to one partition and stay ordered. The broker cluster holds the topic's 200 partitions; each partition has one leader broker that takes all reads and writes and two followers that replicate it, the three forming a replica set. A separate metadata quorum (KRaft in modern Kafka, ZooKeeper in older deployments) is the cluster's brain — it elects partition leaders, tracks which replicas are in sync, and stores configuration, but it never sits on the data path. On the right, independent consumer groups each read the entire stream; within a group, partitions are divided among members so each event is handled once per group.

Components¶

Producer client. Batches events, computes the target partition from the key, and sends to the current leader of that partition. It holds the durability dial (acks) and, when enabled, the idempotence sequence numbers. Producers are dumb about cluster topology beyond a cached metadata map, which they refresh when a leader moves.

Broker. A server that hosts a subset of the cluster's partition replicas on local disk. For partitions it leads, it appends incoming events to the log and serves reads; for partitions it follows, it pulls from the leader to stay in sync. Brokers are nearly stateless in the coordination sense — all cluster state lives in the metadata quorum — so a broker can be replaced by moving its replicas elsewhere.

Partition (the commit log). The unit of ordering, storage, and parallelism. Physically an append-only sequence of segment files with a monotonically increasing offset per event. This is where the data actually lives and the single most important box on the diagram; the LLD opens it.

Replica set & ISR. Each partition is replicated to RF = 3 brokers. The subset currently caught up to the leader is the in-sync replica set (ISR). A write is acknowledged only after all ISR members have it, and only an ISR member is eligible to become leader — this pairing is what delivers "no acknowledged event is ever lost."

Metadata quorum (controller). A small, strongly-consistent Raft group that elects leaders, tracks ISR membership, and holds topic configs. It is consulted on membership changes, not per message, so it stays off the hot path exactly as the ID allocator did in the URL shortener.

Consumer group coordinator. A broker role that assigns partitions to the members of a consumer group and stores each group's committed offsets in an internal compacted topic. When a member joins or dies, the coordinator triggers a rebalance that reassigns partitions so the group keeps making progress.

Primary write path (publish an event)¶

The producer computes partition = hash(key) % 200 for the event's key, so all events for one account target one partition. If no key is set, it round-robins for load spreading.
It looks up the current leader broker for that partition from cached metadata and appends the event to a per-partition batch, flushing when the batch fills or a linger timer expires — batching is what turns a million tiny writes into a few thousand large sequential appends.
The leader appends the batch to the tail of its partition log, assigning each event the next offset, and writes it to the OS page cache (which flushes to disk asynchronously).
The two followers in the ISR pull the batch from the leader and append it to their own logs.
With acks=all, the leader waits until every ISR member has the batch, then returns the base offset to the producer. The event is now durable against the loss of any one broker. With acks=1 the leader acks after its own write only — faster, but a leader crash before replication loses those events.
The producer receives RecordMetadata { partition, offset }. The whole path is a sequential append plus replication, no index maintenance.

Primary read path (consume a stream)¶

A consumer in group warehouse-loader is assigned a set of partitions by the group coordinator. Say it owns P0–P9 of the 200.
It issues a fetch to the leader of each assigned partition, asking for events starting at its current offset — the read is a sequential scan from a known file position, not a query.
The broker serves the bytes straight from the log, ideally via a zero-copy transfer from page cache to socket, so a read never deserializes the events on the broker.
The consumer processes the batch, then commits the offset of the last processed event to the group coordinator. Because the offset is stored by the consumer group, not the broker, a different group reading the same partitions keeps an entirely independent position.
To rewind, the consumer simply seeks to an earlier offset and fetches again — replay costs nothing extra to the broker because the data was never deleted on consumption. Retention, not reading, governs what is still available.

Storage choices¶

Partition log: append-only segment files on local disk. The access pattern is sequential append and sequential scan, which is the one workload spinning and solid-state disks both do fastest, and which needs no B-tree or LSM machinery. Data lives on the broker's own disks, not a shared store, because pushing 3 GB/s over a network filesystem would waste the sequential-IO win. Old segments are deleted or compacted whole, so reclamation is a file unlink, not a row-by-row vacuum.
Consumer offsets: an internal compacted topic (__consumer_offsets). Offsets are themselves a stream of "group X is now at offset Y for partition Z" records; storing them in a log-compacted topic means the latest value per key survives and history is discarded, and it reuses the same replicated-log durability as the data.
Cluster metadata: a Raft-replicated metadata log (KRaft). Leader assignments and ISR membership need strong consistency and are low-volume, so a small Raft quorum fits — the opposite profile from the high-throughput data logs, which is why it is a separate subsystem.

Scaling¶

Write path. Throughput scales by adding partitions and brokers. At 1M/s across 200 partitions each partition carries 5,000 events/s (~5 MB/s), far under a partition's ceiling, so there is headroom to grow to 2M/s on the same partition count. To double capacity you add brokers and reassign partition replicas onto them, spreading leadership so no broker leads a disproportionate share. The partition key must spread evenly — a hot key (one whiplash account doing 100k/s) pins that load to a single partition and no amount of broker-adding relieves it, which is the write-side hot-key trap.

Read path. Reads scale independently of writes because each consumer group reads the whole stream and pays its own bandwidth. Three groups reading live plus one group catching up means the cluster serves roughly 1 GB/s × 3 + 6 GB/s = 9 GB/s of reads against 1 GB/s of ingress — read fan-out, not write rate, often sizes the cluster's NICs and disks. Adding a new consumer group adds read load but never touches the producers.

Consumer parallelism. A group scales up to one consumer per partition — 200 here. Add consumers and the coordinator rebalances partitions onto them; the day-behind loader runs its full 200 to get 6M/s of catch-up throughput. Beyond 200 consumers, extra members sit idle because a partition is single-owner within a group, which is the ceiling that made us choose 200 partitions up front rather than 20.

Operational signals¶

The healthy signal is consumer lag — the offset distance between a partition's newest event and a group's committed offset — sitting near zero and flat; for a live group, lag that stays within a few seconds of production is the system working. The first metric to degrade under trouble is ISR shrink: when a follower falls behind and drops out of the in-sync set, acks=all writes start blocking on the remaining replicas and produce latency climbs, well before any data is lost. The misleading metric is broker CPU, which stays modest because the work is IO and page-cache movement, not computation — a cluster can be melting its disks while CPU looks bored, so watching CPU hides the real bottleneck. The graph an experienced operator opens first during an incident is per-group consumer lag over time: a lag line bending upward tells you a consumer group is losing the race against production, and its slope against the retention window tells you how many hours remain before that group starts losing data.

Failure modes and resilience¶

Broker (leader) failure. When a partition leader dies, the controller elects a new leader from the ISR — a replica guaranteed to have every acknowledged event — and producers refresh metadata and resend to the new leader. Writes pause for that partition for the election window (typically seconds), then resume with zero data loss. Only ISR members are electable, which is the whole point of tracking ISR.
Follower falls behind / ISR shrink. A slow or partitioned follower is removed from the ISR so it cannot block acks; the partition keeps serving from the remaining in-sync replicas at reduced redundancy. When the follower catches back up it rejoins the ISR. The danger knob is min.insync.replicas: set to 2, a partition that shrinks to one in-sync replica refuses further acks=all writes rather than risk acknowledging data held on a single disk.
The day-behind consumer (the threaded scenario). The warehouse loader is offline through an outage and returns 86.4 billion events behind. Because retention is 7 days and its lag is 1 day, every committed offset still points at data on disk, so it resumes exactly where it left off and drains the backlog at ~5M/s of surplus, catching up in ~4.8 hours. Had retention been set to 24 hours to save 1.5 PB of disk, the oldest segments would have aged out from under its offset and it would resume at OffsetOutOfRange — silently skipping a day of transactions. The resilience here is a configuration decision made days earlier, not a runtime lever.
Rebalance storms. When a consumer joins or leaves, the group stops consuming during reassignment ("stop-the-world" rebalance). A flapping consumer can trigger repeated rebalances that stall the whole group. Mitigations: static group membership so a brief restart does not trigger reassignment, and cooperative/incremental rebalancing so only the moved partitions pause.
Unclean leader election. If every ISR member for a partition is lost, the choice is to wait (partition unavailable until an in-sync replica returns) or promote an out-of-sync replica and lose the events it missed. This is a naked availability-versus-durability switch (unclean.leader.election.enable); for a payments stream you leave it off and accept the outage.
Poison message / stuck consumer. One un-processable event at the head of a partition blocks every event behind it, since the consumer cannot advance its offset. Mitigation: a dead-letter topic the consumer routes bad events to so it can commit past them, trading strict order for liveness on that key.

Where this shows up in production¶

LinkedIn — built Kafka to unify activity and operational streams, and runs trillions of messages/day across clusters; the origin case for treating the log as company-wide plumbing rather than one app's queue.
Stripe — routes financial events through Kafka where a duplicate is a correctness bug, the textbook case for the idempotent producer plus transactional exactly-once path rather than plain at-least-once.
Uber — partitions trip and pricing events by geo/rider key so per-entity order holds while the fleet's traffic spreads across partitions, exactly the "order via key, parallelism via partition count" split.
Netflix — runs enormous fan-out where many independent consumer groups read the same event streams, the case that shows read bandwidth, not ingest, sizing the cluster.
Confluent / AWS MSK / Redpanda — package the broker-plus-metadata-quorum architecture as a managed service, and their tiered-storage features are a direct answer to the retention-versus-disk-cost tradeoff by pushing cold segments to object storage.
Datadog / observability pipelines — buffer metric and log floods through Kafka so a slow downstream indexer becomes a growing consumer lag to watch rather than dropped telemetry.
Apache Flink / Spark Streaming — consume Kafka partitions as their source of truth and rely on offset replay for fault-tolerant reprocessing, the reason "reads don't consume" matters for stream processors.
Debezium / change-data-capture — streams database row changes into Kafka keyed by primary key, leaning on per-key ordering so a row's updates arrive in commit order downstream.