Design: Distributed Message Queue (Kafka-like)¶

Difficulty: Senior → Staff | Time: 60–75 minutes

1. Problem Statement¶

Design a distributed message queue like Kafka: producers write messages to named topics, topics are split into partitions for parallelism, and consumer groups read those partitions with tracked offsets.

Every other exercise on this site designs a system that uses infrastructure — a cache, a database, an object store — to serve its own users. This one is different: the system itself is the durability and ordering guarantee that other systems depend on. The notification system and web crawler exercises both assume a queue like this sits underneath them — they are consumers of durable, ordered delivery, not builders of it. If this design silently drops a message, every system built on top of it inherits that bug without knowing it. If this design reorders messages within a partition, any consumer relying on per-key ordering (e.g. "process a user's events in the order they happened") is wrong in a way that's nearly impossible to detect downstream.

That asymmetry — the queue's correctness bugs are invisible until a consumer's business logic breaks weeks later — is what should drive every version below. Do not start with partitioning. Start with: what does "the message is safe" mean, and when are you allowed to say so to the producer?

2. Clarifying Questions¶

3. Functional Requirements¶

• Producers publish messages to a named topic, optionally with a partition key
• Topics are split into partitions; the log within a partition is an ordered append sequence
• Consumers join a named consumer group; partitions are distributed across the group's members
• Consumers track progress via a committed offset per partition, per group
• Messages are retained for a configurable period (or size), independent of whether they've been consumed
• Support replay: a consumer group can reset its offset and re-read history within the retention window

4. Non-Functional Requirements¶

5. Capacity Estimation¶

Producers:
  1M messages/sec sustained, avg message size 1 KB, p99 size 10 KB
  Peak (3x): 3M messages/sec

Ingest bandwidth:
  1M msg/s x 1 KB = ~1 GB/s average, ~3 GB/s peak (this is what the producer
  sends to the leader — logical ingress, one copy)
  With RF=3: leader writes 1 copy locally + sends 2 copies to followers.
  Total physical disk-write volume across the cluster = 3x logical ingress
  ≈ 3 GB/s average, 9 GB/s peak — but "replication traffic" specifically
  (the network bytes the leader sends OUT to followers, separate from the
  producer→leader bytes it already received) is only the 2 follower
  copies ≈ 2x logical ingress ≈ 2 GB/s average, 6 GB/s peak. Don't double-
  count the leader's own local write as "replication" — it isn't network
  traffic, and the 3x figure is total disk-write volume, not network load.

Retention:
  7-day default retention window
  1 GB/s x 86,400 s/day x 7 days ≈ 605 TB of *logical* data
  With RF=3 (3 physical copies): ~1.8 PB of disk across the cluster

Partitioning:
  Target: no single partition exceeds ~10 MB/s of a broker's disk write bandwidth
  1 GB/s / 10 MB/s per partition ≈ 100 partitions minimum, provision 300-500 for headroom and future consumer parallelism

Brokers:
  Each broker: ~20 TB usable disk (multiple disks per node), ~500 MB/s sustained disk write budget
  1.8 PB / 20 TB per broker ≈ 90 brokers minimum just for storage
  9 GB/s peak total disk-write volume (not just network replication —
  every byte written to disk across the cluster, leader + follower copies)
  / 500 MB/s per broker ≈ 18 brokers minimum just for write bandwidth
  Storage is the binding constraint here: a cluster of 90-120 brokers at 20 TB each is a realistic target

Consumer fan-out:
  If 5 independent consumer groups each read the full topic (analytics, billing, audit, search, archival):
  Read bandwidth = 5 x 1 GB/s = 5 GB/s — reads are the multiplier, not writes
  This is why reads should come from OS page cache / follower reads, not force every read to the partition leader's disk

6. API Design¶

# Producer
POST /topics/{topic}/messages
Request:  { "key": "user:123", "value": "<bytes>", "headers": {...} }
Response: { "partition": 4, "offset": 918273, "timestamp": "..." }
  acks=0    → return immediately, no durability guarantee
  acks=1    → return after partition leader writes to its local log
  acks=all  → return after all in-sync replicas acknowledge

# Consumer group membership
POST /consumer-groups/{group}/join
Request:  { "consumer_id": "c-7", "topics": ["orders"] }
Response: { "assigned_partitions": [3, 4], "generation_id": 42 }

# Poll (long-poll style)
GET /consumer-groups/{group}/poll?partition=4&offset=918200&max_wait_ms=500&max_bytes=1MB
Response: { "messages": [ { "offset": 918200, "key": "...", "value": "..." }, ... ] }

# Offset commit
POST /consumer-groups/{group}/commit
Request:  { "partition": 4, "offset": 918273 }

# Admin
POST /topics                         { "name": "orders", "partitions": 200, "replication_factor": 3 }
GET  /topics/{topic}/partitions/{n}/watermark   → { "earliest": 800000, "latest": 918273 }

7. Data Model — the log-structured partition¶

A partition is not a queue data structure — it's an append-only log on disk, split into segment files so old data can be deleted or compacted in whole-file chunks instead of row-by-row.

Partition 4 directory:
  00000000000000000000.log   (offsets 0 - 499,999, sealed, immutable)
  00000000000000000000.index (offset → byte position, sparse)
  00000000000500000000.log   (offsets 500,000 - 999,999, sealed)
  00000000000500000000.index
  00000000001000000000.log   (offsets 1,000,000+, active — currently written to)
  00000000001000000000.index

Each record in a .log file:
  [offset: 8B][length: 4B][crc: 4B][timestamp: 8B][key][value][headers]

Producer append = O(1): fsync to the end of the active segment. No seeks, no random writes.
Consumer read = seek via sparse index to nearest byte offset, then scan forward — O(log n) to locate, O(1) per subsequent read.

Segment rolling: when the active segment hits a size or age threshold (e.g. 1 GB or 7 days), it seals and a new one opens. Retention deletes whole sealed segments once every record in them is older than the retention window — never a row-level delete.

Offset tracking per consumer group lives in a separate internal log (Kafka calls this __consumer_offsets), keyed by (group, topic, partition) → committed_offset, replicated the same way as any other partition — the offset store gets the same durability guarantee as the data it tracks.

8. Version 1 — simplest thing that works¶

Single broker process. Each topic has exactly one partition. Messages live in an in-memory list. No replication.

graph LR
    P[Producers] -->|append| B[Broker process<br/>in-memory list per topic]
    B -->|poll from offset| C[Consumers]

# V1 — single broker, single partition per topic, in-memory
class Broker:
    def __init__(self):
        self.topics: dict[str, list[Message]] = {}

    def produce(self, topic: str, value: bytes) -> int:
        log = self.topics.setdefault(topic, [])
        offset = len(log)
        log.append(Message(offset=offset, value=value, ts=now()))
        return offset  # "acked" the instant it's appended — nothing durable yet

    def poll(self, topic: str, from_offset: int, max_msgs: int = 100) -> list[Message]:
        log = self.topics.get(topic, [])
        return log[from_offset:from_offset + max_msgs]

This handles a single service's event stream fine in a demo. Ship it, then find the actual bottleneck before adding infrastructure.

9. Identify the bottleneck¶

10. Version 2 — partitioning and replication¶

Split each topic into multiple partitions distributed across multiple brokers. Each partition gets a leader (handles all reads/writes for that partition) and N-1 followers replicating it. Producers hash the message key to pick a partition, so all messages for the same key land in the same partition — and therefore keep their order.

graph TD
    Prod[Producers] -->|hash key mod partitions| B1L[Broker 1: P0 leader]
    Prod -->|hash key mod partitions| B2L[Broker 2: P1 leader]
    Prod -->|hash key mod partitions| B3L[Broker 3: P2 leader]
    B1L -->|replicate| B2F[Broker 2: P0 follower]
    B1L -->|replicate| B3F[Broker 3: P0 follower]
    B2L -->|replicate| B1F[Broker 1: P1 follower]
    B2L -->|replicate| B3F2[Broker 3: P1 follower]
    Cons[Consumer group] -->|read from leaders| B1L
    Cons --> B2L
    Cons --> B3L

import zlib

def choose_partition(key: bytes, num_partitions: int) -> int:
    # NOT Python's built-in hash(): CPython randomizes str/bytes hashing
    # per-process by default (PYTHONHASHSEED) for DoS-resistance, so
    # hash(key) % num_partitions would map the SAME key to DIFFERENT
    # partitions across producer restarts or across separate processes —
    # exactly the "same key -> same partition, always" guarantee this
    # function exists to provide would silently break.
    # Use a stable, seed-independent hash instead. CRC32 (used here, via
    # Python's stdlib zlib, purely for a self-contained example with no
    # extra dependency) works fine for this illustration. Kafka's own
    # Java client does NOT default to CRC32 for partition selection —
    # its DefaultPartitioner/StickyPartitioner uses Murmur2 specifically
    # (CRC32 shows up elsewhere in Kafka's protocol, for message
    # checksums, which is an easy mix-up). If you're actually replicating
    # Kafka's exact partition-assignment behavior rather than just
    # building a stable custom hash, use Murmur2 or xxHash, not CRC32.
    return zlib.crc32(key) % num_partitions  # same key -> same partition, always

def produce(topic: str, key: bytes, value: bytes, acks: str = "all") -> int:
    partition = choose_partition(key, num_partitions(topic))
    leader = leader_for(topic, partition)
    offset = leader.append(value)
    if acks == "0":
        return offset                              # fire and forget
    if acks == "1":
        return offset                              # leader wrote to its local log; return now
    # acks == "all": wait for every in-sync replica to confirm
    wait_for_isr_ack(topic, partition, offset)
    return offset

Producer acks trade-off:

For internals on ISR mechanics and how followers catch up, see Kafka Deep Dive and Kafka Internals.

11. Identify the next bottleneck¶

12. Version 3 — production architecture¶

graph TD
    subgraph Producers
        P1[Producer]
        P2[Producer]
    end

    subgraph Cluster["Broker Cluster"]
        B1["Broker 1<br/>P0 leader, P1 follower, P2 follower"]
        B2["Broker 2<br/>P1 leader, P0 follower, P2 follower"]
        B3["Broker 3<br/>P2 leader, P0 follower, P1 follower"]
    end

    subgraph Control["Metadata / Controller Layer"]
        Ctrl["Controller quorum<br/>(KRaft-style, elected via Raft)"]
    end

    subgraph Consumers
        CG["Consumer group coordinator"]
        C1[Consumer 1]
        C2[Consumer 2]
        C3[Consumer 3]
    end

    P1 -->|produce, key-hashed| B1
    P2 -->|produce, key-hashed| B2
    Ctrl -->|leader election, ISR tracking| B1
    Ctrl -->|leader election, ISR tracking| B2
    Ctrl -->|leader election, ISR tracking| B3
    CG -->|assign partitions| C1
    CG -->|assign partitions| C2
    CG -->|assign partitions| C3
    C1 -->|read P0| B1
    C2 -->|read P1| B2
    C3 -->|read P2| B3
    B1 -.->|old segments| Tiered[(Tiered storage<br/>object store)]

Key production decisions:

• Metadata/controller layer. A small quorum (ZooKeeper historically, or a built-in KRaft-style Raft quorum in modern Kafka) tracks partition-to-broker assignment, the ISR set per partition, and runs leader election when a broker dies. The election mechanics themselves — quorum votes, terms, log matching — are exactly Raft; this layer is Raft applied to "who leads this partition," not a new consensus algorithm.
• Consumer group coordinator. One broker per group tracks membership and drives rebalances via heartbeats, deciding which consumer owns which partition and committing that assignment.
• Retention and compaction. Time/size-based retention deletes whole sealed segments; log compaction (for keyed topics like "latest account balance per user") instead retains only the most recent value per key indefinitely, trading history for a bounded, replayable snapshot.
• Rack/AZ-aware replica placement. Followers for a partition are placed in different availability zones than the leader, so a single AZ failure doesn't take out a majority of a partition's replicas at once.

13. Failure analysis¶

14. Consistency considerations — delivery semantics¶

This is the question every interviewer will eventually ask directly: what happens if a message is delivered twice, or not at all?

• At-most-once: consumer commits its offset before processing. If it crashes mid-processing, that message is never retried — it's gone from the consumer's point of view even though the queue still has it. Simple, but silently drops work on crash. Rarely the right default.
• At-least-once: consumer commits its offset after successfully processing. If it crashes between processing and committing, the same message is redelivered on restart — the consumer must tolerate (or dedupe) duplicates. This is the default for this design. It's the only one of the three that never silently loses a message on the consumer side, and it composes with the producer-side acks=all guarantee to give end-to-end "no loss, possible duplicates" — which is a far safer failure mode than "no duplicates, possible loss."
• Exactly-once: achievable only within narrow, deliberately engineered boundaries — an idempotent producer (dedupes retries by sequence number) combined with a transactional read-process-commit-offset that treats the DB write and the offset commit as one atomic unit. It does not extend for free across an arbitrary external side effect (an HTTP call to a third party, for instance) — you cannot make a non-idempotent external effect exactly-once just because the queue promises it internally. See Kafka Internals — Exactly-Once Semantics for the transactional API details.

Design default: at-least-once delivery, with the expectation that consumers implement idempotent processing (dedupe by message id or business key) rather than relying on the queue to guarantee exactly-once end-to-end. This mirrors real production Kafka usage — "exactly-once" is a narrow, opt-in mode, not the ambient default, because most systems have at least one non-transactional side effect somewhere in the pipeline.

15. Observability¶

Metrics:
  produce_latency_p50/p99{acks}
  consumer_lag_messages{group,topic,partition}     ← the single most important metric in the system
  under_replicated_partitions
  isr_shrink_count / isr_expand_count
  broker_disk_used_pct
  rebalance_count{group}
  unclean_leader_elections_total

Alerts:
  consumer_lag > 50% of retention window            (silent data loss risk, see Failure Analysis)
  under_replicated_partitions > 0 for > 5 min        (durability degraded, one more failure = loss)
  rebalance_count{group} > 3 in 10 min               (rebalance storm, see bottleneck section)
  broker_disk_used_pct > 80%
  unclean_leader_elections_total > 0                 (should page — durability was traded for availability)

Traces:
  span from produce() call through leader ack through ISR confirmation, tagged with acks level

16. Cost analysis¶

Broker cluster (100 brokers, 20 TB usable disk each, RF=3):   ~$115,000/month (compute + disk)
Controller quorum (5 small nodes, Raft-based):                 ~$1,500/month
Cross-AZ replication network transfer (RF=3, ~3 GB/s avg):
  3 GB/s × 2.6M sec/month ≈ 7.8M GB/month
  AWS bills BOTH sides of an inter-AZ transfer at ~$0.01/GB each (~$0.02/GB total)
  7.8M GB × $0.02/GB ≈ ~$156,000/month
Monitoring/metrics pipeline:                                   ~$1,500/month
Total:                                                          ~$274,000/month

Cross-AZ replication is now the single largest line item, ahead of the broker fleet itself —
worth calling out explicitly in an interview, since it's the cost that's easy to under-budget
if you only think in terms of "how many brokers do I need for this throughput."

Cost lever: tiered storage (offload segments older than 24h to S3/GCS)
  Local disk needed drops from 7-day retention to ~1-day hot window
  Broker disk requirement drops ~6x → broker count driven by throughput, not storage
  Estimated savings: ~$25,000/month on the storage line — cross-AZ replication cost is
  unaffected by tiered storage, since it's driven by write volume, not retention window

17. Alternative architectures¶

18. Staff Engineer Extensions¶

19. Interview follow-ups¶

1. "Why is ordering only per-partition, not global?" — Global ordering requires all writes to funnel through a single serialization point, which caps write throughput at whatever one node/log can sustain. Per-partition ordering with key-based routing gives you ordering where it matters (per user, per order) while still scaling writes horizontally across partitions. Name the trade-off explicitly rather than treating it as an oversight.
2. "How would a system like the notification service or web crawler use this queue safely?" — They should treat "consumed" as "processed and durably recorded, then committed" — commit-after-processing, not commit-on-read — and build their own idempotency (e.g. dedupe by notification id) because the queue's default is at-least-once, not exactly-once. See notification-system.md and web-crawler.md for where the queue sits in each pipeline.
3. "How do you detect a hot partition and what do you actually do about it?" — Per-partition throughput/lag metrics reveal it (one partition's consumer pinned at 100% CPU while siblings idle). Fixes are application-level: salt the key with a random suffix to spread it across multiple partitions (loses strict ordering for that key, so only where acceptable), or give that one tenant/key a dedicated topic entirely.
4. "Walk through what happens end-to-end if min.insync.replicas=2, RF=3, and the leader crashes right after acking a write." — Tests whether the candidate actually understands ISR: the write was only acked because at least one follower had already confirmed it (2 of 3 replicas), so that follower is a valid failover candidate with the data intact — this is exactly why acks=all + min.insync.replicas>=2 is the durability-safe combination, and why acks=1 alone is not.

Self-Assessment¶

• [ ] I can explain why ordering is per-partition and why that's a deliberate trade, not a limitation
• [ ] I can state the three delivery semantics, this design's default, and why
• [ ] I can walk through unclean leader election as an explicit availability-vs-durability choice
• [ ] I can explain why increasing partition count on a live topic breaks key-based ordering
• [ ] I can name at least two systems on this site that consume a queue like this rather than build one, and what they owe it (idempotency, commit-after-processing)