Design a Distributed Message Queue

A distributed message queue, such as Apache Kafka, Amazon Kinesis, or Google Pub/Sub, sits between services that produce events and services that consume them. The question looks simple: write here, read there. But the moment you add durability, ordering, and millions of events per second, the design space explodes.

Interviewers use this question to probe whether you understand distributed systems fundamentals, not just one vendor's API. The core tensions are: write throughput vs. durability guarantees, strict ordering vs. horizontal scalability, and push-based vs. pull-based consumption. These tensions play out at every level:

The right design depends entirely on scope. A message queue for an internal microservices event bus has very different requirements from a shared platform ingesting click-stream events from 200 million mobile users. Throughput targets, retention windows, and multi-tenancy constraints all change dramatically based on that scope.

Functional requirements

Non-functional requirements

NFR reasoning

Message queues are fundamentally I/O-bound systems. The interesting estimation dimensions are write throughput and storage, not compute. Think in terms of bytes per second first, then convert to broker count and disk size.

A distributed message queue is a durable, ordered log that decouples the services that produce events from the services that consume them. Producers write messages to named topics; consumers pull messages at their own pace using a stored offset. Unlike traditional queues, messages are not deleted after delivery; they are retained for a configurable window, allowing multiple independent consumer groups to replay the same data independently.

The core insight of a log-based message queue is that a durable, ordered, append-only log is both the storage layer and the transport layer. Every message is written once to disk and read many times by many consumers, with zero network copies between storage and consumer in the happy path (a technique called zero-copy, via the OS sendfile() syscall).

Component breakdown

Partition Router (load balancer layer): receives producer requests and routes each message to the correct partition leader. It holds a cached partition-to-broker mapping from the metadata service; on a cache miss or leader change, it refreshes. This layer can be embedded in the producer client library rather than a separate server hop.

Broker cluster: the core of the system. Each broker owns the leadership for a set of partitions and runs the append-to-log cycle. Brokers are symmetric: any broker can be a leader for some partitions and a follower for others. Leadership is assigned by the controller and balances automatically across the cluster.

Metadata service (ZooKeeper / KRaft), tracks which broker is the leader for each partition, which brokers are alive, and what the current ISR (in-sync replica) set is. With KRaft (Kafka's built-in Raft consensus mode), this component is no longer external to Kafka, reducing operational complexity at the cost of requiring a dedicated controller quorum within the cluster.

Append-only log: the actual message store. Each partition is a sequence of segment files on disk. Writes are always sequential appends; reads are random-access by offset but typically sequential as consumers advance through the log. The OS page cache acts as a read buffer, making recent messages essentially free to serve without touching disk.

Offset store (__consumer_offsets): an internal topic that tracks the last committed offset for each consumer group and partition. This is itself a distributed log, giving offset commits the same durability guarantees as regular messages. It also provides the coordination point for consumer rebalancing.

Tiered storage: closed log segments are uploaded to an object store (a durable blob service like S3 or GCS). Brokers only retain the last N days of hot segments locally; consumers requesting older offsets are transparently redirected to fetch from the object store. This is what makes multi-year retention economically viable.

Architectural rationale

Real-world comparison

The central algorithmic question in this design is: given a message, which partition does it go to, and how is that partition physically structured on disk? These two decisions determine throughput, ordering guarantees, and disk efficiency.

Our choice for this design is hash-based partitioning using murmur2 (the same hash Kafka uses). Producers specify an optional partition key; if omitted, the client falls back to round-robin for maximum throughput, useful for workloads where ordering genuinely does not matter.

Log segment structure

Each partition is a sequence of segment files on disk. A segment has a fixed maximum size (default 1 GB) and a maximum age (default 7 days). When a segment is rolled, the old file is closed and becomes immutable, enabling safe concurrent reads and background upload to tiered storage. Active segments are always the last file in the sequence.

Each segment has a companion index file mapping message offsets to byte positions, allowing O(log n) random seeks. The index is sparse, so not every offset is indexed: a seek involves a binary search in the index, then a linear scan of at most a few KB of the log file.

// Producer config: batching + compression
props.put("linger.ms", "10");           // wait 10ms to fill buffer
props.put("batch.size", "65536");     // 64 KB batch
props.put("compression.type", "lz4"); // 3-4x compression on text
props.put("acks", "1");               // leader ack only (at-least-once)

The public API has two core operations: publishing a message and fetching messages. Everything else (offset management, consumer group coordination, partition reassignment) is handled by internal protocols or admin APIs that are unlikely to be the focus of an interview.

POST /topics/{topic}/messages: publish

// Request
POST /topics/orders/messages
Content-Type: application/json

{
  "messages": [
    {
      "key": "user-42",          // optional; determines partition
      "value": "eyJvcmRlcl...",  // base64-encoded payload
      "headers": {              // optional metadata
        "content-type": "application/json",
        "trace-id": "abc-123"
      }
    }
  ],
  "acks": "1"               // "0", "1", or "all"
}

// Response: 200 OK (acks=1 or all) or 202 Accepted (acks=0)
{
  "offsets": [
    { "partition": 3, "offset": 104920 }
  ]
}

// Validation errors
400 Bad Request: message exceeds max.message.bytes (default 1 MB)
429 Too Many Requests: producer quota exceeded (bytes/sec per client)
503 Service Unavailable: not enough in-sync replicas (acks=all, ISR < min.insync.replicas)

GET /topics/{topic}/messages: consume (fetch)

// Request: consumer pulls next batch
GET /topics/orders/messages?partition=3&offset=104921&max_bytes=1048576&wait_ms=500
Authorization: Bearer <consumer-group-token>

// Response: 200 OK
{
  "partition": 3,
  "messages": [
    {
      "offset": 104921,
      "timestamp": 1712700000000,
      "key": "user-42",
      "value": "eyJvcmRlcl...",
      "headers": { ... }
    }
  ],
  "high_watermark": 105000  // latest committed offset in partition
}

// Notable responses
204 No Content: long-poll timeout, no new messages (wait_ms elapsed)
416 Range Not Satisfiable: offset out of retention range (too old)

Optional endpoints by level

The durability guarantee in §2 (no loss once acked) drives every decision in the produce flow. The question is: exactly when does the broker send the ack, and how many replicas must see the message before that happens? The answer creates the central tension between latency and durability.

acks=1 vs acks=all: the key tradeoff

This decision maps directly back to the write latency NFR from §2. With acks=1, the leader receives the message into the OS page cache and acks immediately. A leader crash after ack but before followers replicate loses that message. With acks=all, the leader waits until all in-sync replicas (ISR) have acknowledged before returning, making loss structurally impossible during the ack window, at the cost of latency proportional to the replication round-trip (same AZ: ~10–30 ms; cross-AZ: ~50–100 ms).

There are three categories of entities in this system: messages themselves (the payload), the log structure (how messages are stored and indexed), and the metadata about topics and consumer state. Each is used very differently, and those differences shape where and how each entity is stored.

Access patterns

Two things stand out from this table. First, the message append and fetch patterns are purely sequential I/O: this rules out any key-value store or database as the primary storage layer and points directly to a file-based log. Second, metadata reads happen on every produce and consume request, so they must be served from memory, not from disk.

Message schema (binary, per partition)

Metadata schema (in-memory, replicated via KRaft)

// In-memory partition metadata (per broker)
{
  "topic": "orders",
  "partition": 3,
  "leader_id": 5,           // broker ID
  "isr": [5, 2, 7],         // in-sync replica broker IDs
  "replicas": [5, 2, 7],    // all assigned replicas
  "high_watermark": 104999, // last offset replicated to all ISR
  "log_start_offset": 0,    // first available offset (after retention)
  "epoch": 12               // leader epoch for fencing
}

A Kafka broker does not maintain its own message cache. Instead, it relies entirely on the OS page cache, the kernel's LRU-managed RAM buffer, to serve recent messages. When a consumer fetches messages that are still in the page cache, the broker uses the OS sendfile() syscall to transfer bytes directly from the page cache to the network socket, bypassing the broker process entirely. This is called zero-copy and is why a single broker thread can sustain hundreds of MB/s of consumer throughput at sub-5 ms latency.

A message queue that reads from disk on every consumer fetch would be unusable at scale. The sub-5 ms read latency NFR from §2 is only achievable because recent messages almost never touch disk: the OS page cache absorbs them. Understanding where each cache layer sits in the request path is what separates a good answer from a great one.

Cache hierarchy

Why the OS page cache is the right choice here

Kafka deliberately does not maintain its own in-process message cache. Instead it relies entirely on the OS page cache. This is a deliberate architectural decision: the JVM heap is bounded and subject to garbage collection pauses; the OS page cache is unbounded and managed by the kernel's LRU algorithm, which is well-tuned for sequential access patterns. A broker restart that loses its JVM heap state recovers instantly, because the OS page cache persists across process restarts (until the host reboots).

The consequence is that broker sizing is dominated by RAM, not CPU. A broker with 256 GB RAM and a 48 TB NVMe array will serve warm reads from the page cache at ~2 ms. A broker with 32 GB RAM on the same hardware will see frequent page cache evictions under load, forcing disk reads at ~8–10 ms, a 4–5× latency regression. This is the first thing to check when diagnosing slow consumer reads in production.

Kafka replicates each partition to a configurable number of brokers (replication factor). The broker currently acting as the partition leader accepts all writes; follower brokers continuously fetch and replay the leader's log. The High-Watermark (HWM) is the highest offset that has been confirmed by all in-sync replicas (ISR). Consumers can only read up to the HWM. Messages above it exist on the leader but have not been replicated to all ISR members yet. This guarantee means a consumer never reads a message that could disappear after a leader failover.

Durability in this system is a property of the replication flow, not of disk writes alone. A single disk write survives a process crash but not a disk failure. Replication across brokers (ideally across availability zones) is what makes data durable against hardware failures.

Leader election and log truncation

When a new leader is elected, it issues a LeaderEpochRequest to all follower replicas. Any follower whose log-end-offset (LEO) exceeds the new leader's committed high-watermark for that epoch must truncate the excess entries. This log truncation step is what enforces the HWM contract: no consumer ever reads a message that was written above the HWM, because followers discard those writes when they learn of the new leader's epoch. Without this truncation, a deposed leader that comes back online and a newly-elected leader could have diverged logs, potentially creating duplicate or inconsistent consumer reads.

The leader epoch (the epoch field in the metadata schema in §7) is a monotonically increasing counter that prevents split-brain: followers reject writes from any broker whose epoch is lower than the controller's current assignment. An old leader that recovers from a network partition detects the stale epoch and stops accepting writes before rejoining as a follower.

Exactly-once delivery: how it actually works

Scaling a message queue is fundamentally about scaling partitions and scaling disk. Compute is rarely the bottleneck: network bandwidth and disk I/O are. The diagram below shows a production-grade multi-region setup.

// Set quota for client "order-service" via Admin API
kafka-configs --bootstrap-server broker:9092 --alter \
  --add-config 'producer_byte_rate=10485760' \  // 10 MB/s
  --entity-type clients --entity-name order-service

Classic probes table