Design a Kafka-like distributed message queue

Source: https://www.hack2hire.com/question-bank/questions/system-design/69ebc33b068f42e1550890b8
Company: Databricks, Microsoft
Difficulty: Hard

Problem statement

Design a durable, partitioned log that producers append to and many consumer groups read at their own pace, surviving broker failures without losing acknowledged messages.

Requirements

We are building a Kafka-style log service: producers append records to partitioned, replicated topics; many independent consumer groups read the same data at their own pace and commit how far they got; a broker crash never loses a write that was acknowledged with acks=all.

Clarify first: ordering scope (per partition only, not per topic), retention (time and size, not "until consumed"), delivery semantics (at-least-once by default, exactly-once as an opt-in), and whether multi-region is synchronous (no: async mirroring).

Functional requirements

• Create topics with N partitions and replication factor RF; records are (key, value, headers, timestamp).
• Produce: records with the same key land on the same partition and keep their order there.
• Fetch: a consumer reads a partition from any offset, sequentially, in batches.
• Consumer groups: each partition is owned by exactly one member of a group; groups are independent of each other.
• Commit and fetch offsets per (group, topic, partition) so a restarted consumer resumes where it left off.
• Retention by time (e.g. 7 days) or size, plus log compaction (keep the latest value per key) for changelog topics.

Non-functional requirements

• Durability: an acks=all write survives the loss of RF−1 replicas. Never lose a committed record.
• Availability: a partition elects a new leader within a few seconds of a broker failure.
• Throughput first: millions of records/s per cluster; p99 produce latency under ~20 ms within a region.
• Horizontal scale by adding partitions and brokers; a consumer is never slower because another group is reading.
• Ordering is guaranteed only within a partition.

Back-of-the-envelope estimation

• Ingest 2M records/s × 1 KB = 2 GB/s. With RF=3, disks write 6 GB/s across the cluster; replication adds 4 GB/s of broker-to-broker network.
• Fan-out: 5 consumer groups read everything → 10 GB/s egress, mostly served from page cache because consumers are near the tail.
• Retention 7 days: 2 GB/s × 604,800 s ≈ 1.2 PB raw, 3.6 PB with RF=3. That argues for tiered storage (see below).
• A broker with 12 × 16 TB disks and 25 Gbps NIC sustains ~300 MB/s ingest including replica traffic and consumer egress → ~20–30 brokers for throughput, more for disk if everything stays local.
• Partitions: target ~10 MB/s per partition → 200+ partitions for the main topic; keep under ~4,000 partition replicas per broker so failover metadata stays small.

API design

A binary request/response protocol over long-lived TCP connections, versioned per request type. Clients pipeline many in-flight requests per connection. All data calls go to the partition leader; clients learn leaders from metadata.

• Metadata(topics[]) → {brokers[], topics[{name, partitions[{id, leader, replicas[], isr[], leader_epoch}]}]}. Clients cache this and refresh on NOT_LEADER errors.
• Produce(acks: 0|1|all, timeout_ms, transactional_id?, batches[{topic, partition, record_batch}]) → [{topic, partition, base_offset, log_append_time, error}]. A record batch carries producer_id, producer_epoch, base_sequence, compression codec, and a CRC.
• Fetch(replica_id, max_wait_ms, min_bytes, max_bytes, isolation_level, partitions[{topic, partition, fetch_offset, leader_epoch}]) → [{partition, high_watermark, last_stable_offset, records, error}]. This is long-poll: the broker parks the request until min_bytes are available or max_wait_ms passes. Followers use the same call with their replica_id.
• OffsetCommit(group_id, generation_id, member_id, offsets[{topic, partition, offset, metadata}]) → errors[]; OffsetFetch(group_id, partitions[]) → [{partition, offset}]. A stale generation_id is rejected, which fences zombie consumers.
• Group protocol: FindCoordinator(group_id), JoinGroup, SyncGroup, Heartbeat, LeaveGroup.
• Supporting: ListOffsets(partition, timestamp) for seek-by-time; CreateTopics, DeleteRecords; InitProducerId, AddPartitionsToTxn, EndTxn for transactions.

Data model

Partition log on broker disk. Each partition replica is a directory of segments: 00000000000052340000.log (record batches appended in order), a sparse .index (offset → byte position, one entry every ~4 KB), and a .timeindex (timestamp → offset). Segments roll at 1 GB or 1 hour. Retention deletes whole segments; compaction rewrites closed segments keeping the last record per key. Plain files on local disk, because the access pattern is pure append plus sequential read, which the OS page cache handles better than any database.

Per-partition state (in memory, checkpointed to files): log_end_offset (LEO), high_watermark (HW, the highest offset replicated to all ISR members), leader_epoch plus an epoch → start-offset cache used for truncation after failover.

Consumer offsets. Stored in an internal compacted topic __consumer_offsets, 50 partitions, keyed by (group_id, topic, partition) with value {offset, leader_epoch, metadata, commit_ts}. The partition is hash(group_id) % 50, and the leader of that partition is the group's coordinator. Commits are just produces, so offsets get the same replication and durability as data; the coordinator keeps the latest value per key in a hash map for fast OffsetFetch.

Cluster metadata. A replicated metadata log run by a small controller quorum (3 or 5 nodes, Raft): topics, partition → replicas, current leader, ISR, leader epoch, broker registrations. Brokers tail this log and keep a local copy, so the controller never has to push full state.

Producer state. Per partition: producer_id → {epoch, last 5 batch sequence numbers, last offset} for dedup, snapshotted with segments.

High-level design

flowchart LR
  P[Producers] -->|Produce| L1[Broker 1 leader P0]
  L1 -->|follower fetch| F2[Broker 2 follower P0]
  L1 -->|follower fetch| F3[Broker 3 follower P0]
  CG1[Consumer group A] -->|Fetch| L1
  CG2[Consumer group B] -->|Fetch| L1
  CG1 -->|OffsetCommit| CO[Group coordinator]
  CO --> OT[(Offsets topic)]
  CQ[Controller quorum] -. metadata log .-> L1
  CQ -. metadata log .-> F2
  CQ -. metadata log .-> F3
  L1 -->|cold segments| TS[(Object storage)]

1. Produce with acks=all. The producer batches records per partition (linger 5 ms, 64 KB), compresses the batch, and sends it to the leader. The leader validates sequence numbers, appends to its log (page cache, no fsync), and parks the request in a purgatory. Followers continuously Fetch from the leader; each fetch at offset X tells the leader that the follower has everything below X. When every ISR member's fetch offset passes the batch end, the leader advances HW and answers the producer with base_offset. If ISR size drops below min.insync.replicas (2 with RF=3), the leader rejects the write with NOT_ENOUGH_REPLICAS instead of acknowledging it with weaker durability.
2. Consumer fetch and commit. A group member finds its coordinator, joins, and gets assigned partitions. It looks up committed offsets, then long-polls Fetch on each partition leader. Consumers only see records below HW, so they never read something that could be lost on failover. After processing, it commits offsets (every 5 s or after each batch); a crash replays from the last commit, which is at-least-once.
3. Leader failover. The controller sees broker 1's session expire (no heartbeat for ~6–10 s). For each partition it led, the controller picks the first live replica in the ISR, bumps leader_epoch, and writes the change to the metadata log. Brokers apply it; the new leader starts serving; clients get NOT_LEADER_OR_FOLLOWER, refresh metadata, and retry. Producer retries are deduplicated by sequence number, so retries do not create duplicates.

Why it holds together: the log is the single abstraction. Data, replication (followers are just consumers), offsets (a topic), and metadata (a Raft log) all reuse it, so one durability argument covers everything.

Deep dives

Replication correctness: ISR, HW, and leader epochs.

• The ISR is the set of replicas within replica.lag.time.max.ms (e.g. 30 s) of the leader's LEO. A slow follower is removed from the ISR via the controller, so one bad disk does not stall every write; it rejoins after catching up.
• A record is committed when it is below the HW. With acks=all and min.insync.replicas=2, every acknowledged record exists on at least 2 brokers, and only ISR members can become leader, so the new leader always holds every committed record.
• No fsync per write: durability comes from replication across failure domains (rack-aware placement puts the 3 replicas in 3 AZs), not from one disk. A simultaneous power loss in all AZs is the accepted risk; latency-sensitive teams can opt into fsync.
• Truncation after failover: an old leader that comes back may have uncommitted tail records the new leader never got. It asks the new leader "what is the end offset of my last epoch?" (OffsetsForLeaderEpoch) and truncates to that point before fetching. Using the epoch instead of the old HW avoids the classic bug where two replicas diverge after back-to-back failures.
• Unclean leader election (picking an out-of-sync replica when the whole ISR is gone) is off by default: it trades lost data for availability. Some telemetry topics turn it on.

Consumer group rebalancing.

• The coordinator runs a generation-numbered protocol: members JoinGroup, one member (the leader) computes the assignment with a pluggable strategy (range, round-robin, sticky), and SyncGroup distributes it. Members heartbeat every 3 s; a missed session (45 s) or a poll gap over max.poll.interval.ms triggers a rebalance.
• Eager rebalancing revokes all partitions from everyone, which stops the whole group for seconds; with 500 members that is visible lag. Cooperative (incremental) rebalancing revokes only the partitions that move, in two rounds, so the rest keep consuming.
• Static membership (group.instance.id) lets a pod restart within the session timeout and get its old partitions back without a rebalance; rolling deploys stop causing churn.
• Fencing: commits carry generation_id; a consumer that was kicked out (GC pause) gets ILLEGAL_GENERATION and cannot overwrite the new owner's offsets.
• Parallelism is capped by partition count: 200 partitions means at most 200 active consumers per group. Choose partitions for expected peak consumers, since adding partitions later changes key → partition mapping.

End-to-end throughput: batching, page cache, zero-copy.

• Batching at every layer: producers send compressed batches (zstd, ~4× on JSON); the broker stores the batch as is and never decompresses it on the hot path (it only checks the CRC and assigns offsets); consumers receive the same bytes. One 64 KB batch replaces ~64 syscalls and headers.
• Sequential I/O: appends go to the end of one active segment per partition, so disks see large sequential writes (~200 MB/s per HDD, far more on NVMe).
• Page cache instead of an application cache: the broker keeps almost no heap (6 GB) and leaves the rest of RAM (e.g. 128 GB) to the OS. Consumers near the tail hit memory; replicas and consumers share the same cached pages. A restart does not lose a warm cache.
• Zero-copy: for plaintext connections, sendfile moves bytes from page cache to the socket without copying into user space. TLS breaks this (encryption needs user-space buffers), so expect 20–30% more CPU with TLS.
• Lagging consumers reading old data cause disk reads that evict hot pages; cap their fetch rate with quotas, or serve them from tiered storage so they do not hurt tail readers.

Effectively-once: idempotent producers and transactions.

• Idempotence: InitProducerId gives each producer a producer_id and epoch. Each batch has a per-partition sequence number; the leader keeps the last 5 per producer and drops duplicates, rejecting gaps with OUT_OF_ORDER_SEQUENCE. Retries become safe, and this state is replicated with the log so it survives failover.
• Transactions: a transactional_id maps to a transaction coordinator (a partition of __transaction_state). The producer registers partitions, writes records, adds the consumer offsets to the same transaction, then EndTxn. The coordinator writes PREPARE, then commit markers into every touched partition, then COMPLETE. Read-process-write pipelines thus commit output and input offsets atomically.
• Consumers with read_committed read only up to the last stable offset (LSO, the first offset of the oldest open transaction) and skip aborted records using an aborted-transaction index. A stuck transaction holds back the LSO, so transactions have a timeout (e.g. 60 s).
• A restarted producer with the same transactional_id bumps the epoch, which fences the zombie instance. Side effects outside the log (an email, a DB write) are not covered; those still need idempotent sinks.

Tiered storage and multi-region.

• Tiered storage: closed segments are copied to object storage (S3/GCS) with their indexes; brokers keep only ~1 day locally. Retention can be months at ~$0.02/GB-month instead of 3× replicated SSD. Fetches below the local log start read from the remote tier through a separate thread pool and cache. Broker replacement gets fast because a new replica copies only the local hot tail.
• Multi-region: replicating synchronously across regions would add 50–100 ms to every acks=all write and make a WAN partition stop writes. Instead, run a cluster per region and mirror topics asynchronously with a replicator (consumer + producer) that also translates consumer offsets. RPO is the mirror lag (seconds); on regional failover consumers resume from translated offsets and must tolerate some duplicates.

Health signals. Under-replicated partitions (should be 0); offline partitions (page immediately); ISR shrink/expand rate; produce p99 and request-queue time; consumer lag in records and seconds per group; controller event queue and metadata lag; disk usage per broker and partition skew; network processor idle ratio below 30% means saturation.

Trade-offs

• Pull-based consumers make backpressure and replay trivial and let each group go at its own pace, but add up to max_wait_ms latency and need long-polling to avoid busy loops.
• Per-partition ordering only: global order would cap throughput at one leader. Teams that need ordering must pick keys carefully, and hot keys create hot partitions that cannot be split without breaking order.
• Replication without fsync gives high throughput, but correctness depends on replicas failing independently; correlated power loss across all AZs can lose acknowledged data.
• min.insync.replicas=2 chooses consistency over availability: with two of three replicas down, the partition rejects writes rather than accepting them unsafely.
• ISR (leader waits for all in-sync followers) keeps the replica count low (RF=3 tolerates 2 failures for reads) but the slowest ISR member sets latency; a quorum write (majority acks) is tail-tolerant but needs 2f+1 replicas to tolerate f failures.
• Retention-based deletion keeps brokers simple and enables replay, but storage cost scales with retention, not with consumption; tiered storage moves cost to object storage at the price of slower reads of old data.
• Consumer offsets in a compacted topic reuse the log's durability, but committing very often (every record) creates heavy write load on __consumer_offsets.
• Async cross-region mirroring keeps regional writes fast but gives a non-zero RPO and offset translation complexity on failover.

Interviewer expectations

Mid-level (L4)

• Explains topics, partitions, offsets, and consumer groups, and why ordering is per partition.
• Designs produce and fetch APIs with batching and a key-hash partitioner.
• Stores consumer offsets durably and describes at-least-once delivery with commit-after-process.

Senior (L5)

• Gets replication right: ISR, high watermark, acks=all with min.insync.replicas, and why consumers only read below HW.
• Walks through leader failover end to end, including client metadata refresh and producer retries.
• Explains consumer rebalancing, its cost, and how cooperative or static membership reduces it.
• Uses numbers to size partitions, brokers, disks, and network, and justifies page cache plus zero-copy.

Staff (L6+)

• Covers log truncation with leader epochs and the divergence it prevents; knows when unclean election is acceptable.
• Designs idempotent producers and transactions (sequence numbers, epochs for fencing, LSO for read_committed) and states their limits.
• Separates the metadata plane (Raft controller quorum) from the data plane and discusses partition count limits.
• Plans for operations: tiered storage, async multi-region mirroring with RPO, quotas for noisy tenants, and the health signals that page on-call.