System Design Interview: Design a Distributed Message Queue

Problem and Applicable Scenarios

Design a distributed message queue shared by multiple product teams. Steady ingress is 1 million messages per second at an average of 1 KB each, and a traffic peak can sustain 3 million messages per second for 15 minutes. Messages are retained for 24 hours by default. Producers publish in batches. Consumers pull as groups, commit offsets, and replay within retention. Messages with the same business key require local ordering, while different keys may run in parallel. Ordinary messages use at-least-once delivery, and publish acknowledgment has a p99 target under 50 milliseconds. Each partition has three replicas across three availability zones; losing one zone must not lose acknowledged messages.

Most messages are small, but the API permits business payloads up to 100 MB. A large payload should not repeatedly pass through broker logs, replica transfers, and consumer buffers. In this design it first enters object storage, while the queue stores only an immutable reference, size, and checksum. Throughput, latency, retention, thresholds, and replica counts are interview assumptions that require benchmarking on the target hardware. They are not product claims.

A Chinese system-design article published in May 2026 directly discusses a queue processing tens of billions of messages per day with million-QPS peaks. A PracHub prompt updated in June 2026 asks candidates to cover APIs, consumer groups, offsets, partitions, replicas, large payloads, and multi-tenant isolation. Together they establish a currently verifiable system-design prompt. Only one page asserts a company attribution, so this article leaves the company unset.

What Interviewers Evaluate

First, does the candidate define the delivery contract? A successful publish, durable broker storage, message delivery to a consumer, completion of a business side effect, and an offset commit are five separate boundaries. Calling all of them “message success” hides two failure windows: a producer may repeat a publish after losing the acknowledgment, and a consumer may repeat work after completing the side effect but crashing before the offset commit.

Second, do partitioning, ordering, and scaling follow one argument? Sending one key to one partition preserves that key's log order. The same partition limits both write throughput and consumer-group parallelism. More partitions add parallel slots, but they do not split a continuously overloaded hot key or create global ordering.

Third, does the replica acknowledgment rule survive failure? A strong answer states when the producer can be acknowledged, which replicas are eligible to become a new leader, and how an epoch fences a recovered old leader. Saying “three replicas fail over automatically” does not prove that acknowledged records survive.

Fourth, are consumer offsets connected to business results? Committing before processing can skip a business effect. Processing before committing can replay it after a crash. At-least-once chooses the second window, then absorbs repeats with a stable message ID, business idempotency key, unique constraint, or version condition. A broker transaction covers only resources participating in that transaction; it does not automatically grant exactly-once effects to an external payment service, email provider, or database.

Finally, can the queue preserve these boundaries under pressure? The candidate should calculate 24-hour storage and burst backlog, handle large messages, hot partitions, slow consumers, poison messages, full disks, and noisy tenants, then inject failures to prove that acknowledged messages are not lost, uncommitted work is not skipped, and stale owners cannot advance offsets.

Clarifying Questions Before Answering

• Is this a retained log or a claim-and-delete work queue? This design chooses a retained log because multiple

consumer groups and replay are required. If exactly one worker claims each job and historical replay is unnecessary, a lease queue with visibility timeouts is simpler.

• What is the ordering scope? Ordering covers append order within one topic partition, with one business key pinned

to one partition. There is no global order across keys, partitions, or topics. Global order would reduce throughput to one serial log.

• What does publish acknowledgment mean? Two of three replicas have durably stored the record, and the control plane

still recognizes the current leader epoch. A strongly durable topic rejects writes when fewer than two replicas are available.

• What is the consumption semantic? The default is at least once: process successfully, then commit the next offset.

External side effects require idempotency or reconciliation. Only a low-value workload that permits loss but forbids duplicates should commit first.

• Do consumers need arbitrary replay? A group can reset by offset or timestamp within the 24-hour retention window.

Beyond retention it must restore from an archive, or fail explicitly when no archive exists.

• May poison messages be skipped? Ordinary topics can move a message to a dead-letter topic after bounded retries.

A strictly key-ordered topic cannot skip it for free; pause the key or partition and repair it, or later messages may pass the failure.

• Must a 100 MB payload be inline? No. This design assumes inline payloads up to 256 KiB and uses object references

above that threshold. Benchmarking and cost determine the threshold; 100 MB is the object-payload limit.

• What is the cross-region requirement? The primary design is one region across three availability zones.

Asynchronous cross-region disaster recovery cannot promise both zero data loss and local write latency. A zero-RPO cross-region requirement changes the acknowledgment path and latency budget.

• How strong is tenant isolation? Brokers are shared by default, with per-tenant ingress, egress, storage, partition,

and connection limits. Very large or regulated tenants may use a dedicated broker pool under the same control plane and protocol.

30-Second Answer Framework

“I would model this as a retained, replayable partitioned log. The producer routes by business key to a partition leader, batches writes, and receives an acknowledgment only after two replicas in different availability zones persist the batch. That gives per-key order, not global order. A consumer group exclusively owns partitions, completes an idempotent business write, and then commits the next offset, so delivery is at least once and external effects deduplicate by message ID. Steady ingress is about 1 GB/s and 86.4 TB per day logically, or 259.2 TB with three replicas. A 15-minute three-times peak creates about 1.8 TB of extra backlog if consumers sustain the steady rate. Payloads above 256 KiB go to object storage and the queue carries a reference and checksum. I would prove the boundaries with tenant quotas, fair scheduling, lag alerts, leader failure, lost acknowledgment, and consumer-crash tests.”

Step-by-Step Deep Dive

Step 1: Write the APIs and invariants before the components.

The essential surface covers topics, publishing, fetching, committing, and resetting offsets:

POST /v1/topics
POST /v1/topics/{topic}/messages:publish
POST /v1/groups/{group}/messages:fetch
POST /v1/groups/{group}/offsets:commit
POST /v1/groups/{group}/offsets:reset

A publish request carries an authenticated tenant context, topic, optional message_key, stable message_id, payload or object reference, producer epoch, and per-partition sequence. The batch response returns each message's partition, offset, and commit status. Fetch carries the group, partition-ownership generation, starting offset, maximum bytes, and long-poll duration. Commit writes the offset of the next record to read.

The design preserves four invariants: an acknowledged record remains readable after one availability-zone failure; consumers see only a committed prefix; offsets increase monotonically within a leader epoch; and a consumer with an expired generation cannot commit offsets or continue writing results. The API-level message_id supports business deduplication. producer_id + epoch + sequence lets the broker recognize a retry of the same publish.

Step 2: Separate the control plane from the data plane.

Control plane: tenants and ACLs, topic configuration, partition placement,
               replica membership, leader epochs, quotas

Data plane:
Producer -> metadata cache -> partition leader -> follower replicas
Consumer group -> group coordinator -> partition leaders -> business sink

A small consensus-backed cluster stores topic and partition metadata and assigns a monotonically increasing epoch to each leader term. It does not carry message bodies. Brokers append, replicate, read, and retain records. Clients cache partition leaders and refresh metadata after a stale-epoch or not-leader response. Message throughput avoids a central proxy, while existing partitions can continue for a bounded lease during a control-plane outage. Topic creation and partition movement may pause; stale metadata must never elect an arbitrary leader.

Step 3: Use partitioned logs for throughput, replay, and local ordering.

Each partition is a set of append-only segments whose records contain:

MessageEnvelope {
  tenant_id, topic, partition, offset
  message_id, message_key, producer_id, producer_epoch, sequence
  created_at, headers, payload_or_ref, payload_size, checksum
}

The active segment receives sequential appends. A sparse offset index locates reads, and closed segments roll by time or size. Consumers fetch batches by offset, allowing sequential I/O, page-cache use, and batched network transfer. Retention deletes whole segments after 24 hours. A segment under valid replay or tiered-storage upload holds a reference so deletion cannot race with a reader.

The routing hash includes the trusted tenant, topic, and business key. Records with the same key stay on one partition; unkeyed records may use round-robin or sticky-batch assignment. Tenant-only routing makes a large tenant hot, while fully random routing loses key order. Adding partitions affects future records. A changed modulo can place one key on both old and new partitions. When stable order matters, map virtual shards to physical partitions; pause the virtual shard during movement, record the cutover offset, drain the old owner, and resume under a new epoch.

Step 4: Give replication and leader election one definition of commit.

Each partition has three replicas in three availability zones. The leader assigns offsets, durably appends a batch, and replicates it in parallel. Once any two replicas have persisted the batch, commit_watermark advances and the producer is acknowledged. Consumers read only offsets below that watermark. On leader failure, the control plane selects only a replica containing the committed prefix and increments the epoch. A recovered old leader truncates its uncommitted tail and catches up before serving; requests carrying its old epoch are rejected.

This policy tolerates one availability-zone failure. With two replicas left, both must acknowledge, so latency and capacity degrade. If either remaining replica then fails, a strongly durable topic stops acknowledging writes until replication is restored. Electing a stale replica to improve availability would violate the no-loss promise. During a network partition, only the side with a commit majority may write; the other side is fenced.

A producer retries when the record committed but the acknowledgment response was lost. The broker deduplicates within the partition using producer epoch and monotonic sequence and returns the original offset. A zombie producer with an old epoch is rejected. This removes log duplicates caused by publish retries. It does not combine two distinct business requests that used different IDs, or deduplicate a consumer's external side effect.

Step 5: Connect group ownership and offsets to the business result.

Within a consumer group, one member owns a partition at a time. The group coordinator maintains members, leases, generations, and assignments. A timeout or scaling event creates a new generation, and fetch or commit from an old member is rejected. Incremental rebalancing moves only necessary partitions and reduces group-wide pauses, but a consumer must still stop fetching and commit finished work before ownership is revoked.

The default order is fetch a batch, perform an idempotent business write, then commit the next offset. If the consumer crashes after the business commit and before the offset commit, the new owner replays completed records, which creates at-least-once delivery. A stable message_id can back a unique constraint or processed-message record, or be committed in the same database transaction as business state. If output returns to the same messaging system, the output records and input offset may share a broker transaction. An external database, payment, or email provider still needs idempotency, status lookup, or reconciliation.

Offsets are stored in a replicated metadata log under (tenant, group, topic, partition) and carry the generation. Monitoring includes both log_end_offset - committed_offset and the age of the oldest unprocessed message. Message count alone misstates a backlog with variable record sizes, so the system also reports lag bytes and catch-up time at the current net consumption rate.

Step 6: State the conflict among retries, dead letters, and ordering.

Transient network and throttling failures enter a delayed retry topic with jitter. Deterministic schema, permission, or business-validation failures should not be retried blindly. A retry preserves the original message_id, source topic, partition, offset, first-seen time, attempt count, and error class. After the attempt or business-deadline limit, it moves to a dead-letter topic, raises an alert, and permits controlled redrive. Redrive keeps the original ID so it cannot bypass idempotency.

Moving a failed record aside allows later records to finish first, which conflicts with strict per-key order. If order state must evolve strictly, pause that key and buffer its later records in a separate ordered lane, then resume from the failed offset after repair. Pausing the entire partition is simpler but has a larger blast radius. If the business accepts convergence by version, later records can proceed and the sink rejects stale versions. The topic contract must choose; it cannot promise both “poison messages never block” and “messages never pass one another.”

Step 7: Put large messages behind object references and close the garbage-collection races.

This design sets 256 KiB as the inline threshold. A larger payload uses short-lived upload credentials to write an immutable object with its size, content hash, and encryption metadata. Only after the upload succeeds does the producer publish the reference. A consumer reads the object and verifies the hash. Broker replicas copy only the small envelope, so a 100 MB record cannot monopolize network buffers, replication batches, or consumer memory.

A successful upload whose reference was never published is an orphan and expires with the upload-session TTL. Once the reference commits, object retention must cover message retention, valid replay, dead letters, and a safety margin. A deletion job first checks protected references and deletes after a grace period. When object reads fail, consumption remains uncommitted and retries; acknowledging first could leave a permanently missing body. Large payloads receive separate per-tenant byte-rate, concurrent-download, and storage quotas because message-count throttling misprices them.

Step 8: Derive partitions, disk, and catch-up headroom from capacity.

Using decimal 1 KB, steady logical ingress is:

1,000,000 messages/s × 1,000 bytes = 1 GB/s
1 GB/s × 86,400 s = 86.4 TB/day
Lower bound for three replica writes = 86.4 × 3 = 259.2 TB/day

Total ingress during the 15-minute peak is 3 GB/s × 900 = 2.7 TB. If consumers sustain only the steady 1 GB/s, the extra backlog is:

(3 GB/s - 1 GB/s) × 900 s = 1.8 TB

After the peak, suppose consumers sustain 1.5 GB/s while new ingress remains 1 GB/s. The net catch-up rate is 0.5 GB/s, so 1.8 TB takes 3,600 seconds, or about one hour, to drain in theory. Replica recovery, batch overhead, compression, indexes, filesystem reserve, and large-message object storage add capacity, so these are lower bounds.

Partition count is bounded by both bytes and messages. Suppose a benchmark with three replicas and the target p99 finds that one partition sustains 40 MB/s and 40,000 messages per second. Both peak dimensions require at least 75 partitions. Adding 50% headroom for failure and rebalancing yields about 113, so 128 is a practical choice. That per-partition result is an interview benchmark assumption. Different hardware, batches, acknowledgments, or record sizes require a new test; 128 is not a universal answer.

Step 9: Implement backpressure, tenant isolation, and verifiable operations.

Consumers use long polling and control their rate with max_bytes and in-flight batch limits. As broker disk watermarks rise, the system first stops new partition creation, lowers burst allowances for low-priority tenants, then rejects over-quota publishes with a retryable signal. Unbounded memory queues and retries turn congestion into process failure. Producers use bounded batch buffers, deadlines, and jittered backoff so a broker outage cannot create synchronized retry storms.

Tenant identity comes from credentials, never the message body. Ingress is limited by message and byte rate. Egress is fair-scheduled by fetch bytes and request CPU. Storage, partitions, consumer groups, connections, in-flight requests, and large objects also have limits. Placement avoids concentrating one tenant's replicas or hot partitions on a few brokers. Very large tenants move to dedicated pools, while shared pools still meter and report rejections per tenant.

Key metrics include publish acknowledgment p50/p95/p99, errors and unknown outcomes; per-partition ingress bytes, leader and follower lag, commit_watermark, disk watermark, and hot keys; consumer-group committed offsets, consumer lag, oldest age, rebalances, retries, and dead letters; object orphans and read failures; and per-tenant throttling and fairness. An end-to-end canary publishes a stable ID, commits an idempotent business result, then commits its offset and reconciles broker, group, and business states.

The fault matrix includes a leader crash before replication, after commit, and before returning the acknowledgment; one availability-zone loss; network partition; full disk; stale-leader recovery; hot key; a 15-minute three-times peak; consumer crashes before and after its business commit; stale commits during rebalance; poison messages; upload success followed by publish failure; object-read failure; and dead-letter redrive. Acceptance asserts that acknowledged records survive, uncommitted work is not skipped, key order follows the topic contract, stale generations cannot advance offsets, and every duplicate, rejection, or drop has an attributable metric.

High-Quality Sample Answer

“I would first confirm that this is a retained-log service requiring multiple consumer groups and 24-hour replay. A topic is split into partitions, and the same business key stays on one partition. Ordering therefore covers a key and partition, while different partitions run in parallel. Producers obtain leader metadata from the control plane and write batches directly. Each partition has three replicas across availability zones; only two durable replicas advance commit_watermark and acknowledge the producer. A new leader must contain the committed prefix, and epochs fence old leaders and producers.

Storage uses append-only segments and a sparse offset index. Group consumers exclusively own partitions and long-poll. The consumer commits an idempotent business result before committing the next offset. A crash can replay work but cannot silently skip it. On the producer side, producer epoch and sequence remove retries caused by a lost acknowledgment. On the consumer side, a stable message ID, unique constraint, or version condition absorbs repeated effects. I would claim an end-to-end broker transaction only when both input offset and output live in that broker transaction; external systems still need idempotency or reconciliation.

Steady capacity is 1 GB/s and 86.4 TB of logical data per day, with a lower bound of 259.2 TB for three replica writes. A three-times peak for 15 minutes creates 1.8 TB of extra backlog when consumer capacity stays at steady state. If the post-peak net catch-up rate is 0.5 GB/s, draining it takes about one hour in theory. Partition count uses the larger of the message-rate and byte-rate calculations, adds failure headroom, and is calibrated on the real hardware.

Payload bodies above 256 KiB first enter object storage. The queue retains an immutable reference, size, and checksum. Upload-session TTL removes orphans, while a committed reference protects its object through retention, replay, and dead-letter windows. Retries preserve the original message ID. A strictly ordered topic pauses a key or partition on a poison message because immediately dead-lettering it would let later messages pass.

Finally, I would limit ingress, egress, storage, partitions, and large objects per tenant; isolate a hot tenant in a dedicated pool; and monitor acknowledgment latency, unknown publishes, replica lag, disk watermarks, oldest consumer age, hot keys, and dead letters. Fault tests cover lost acknowledgments, leader crashes before and after commit, zone failure, stale offset commits, consumer crashes after business writes, and object-storage failure. Each test checks the specific acknowledgment boundary.”

Common Mistakes

• **Mistake: Draw only Producer, Kafka, and Consumer → Failure: Component names do not define acknowledgment, offset,

order, or failure boundaries → Fix: State the delivery contract and four invariants, then map every component to one.**

• **Mistake: Promise global order while scaling horizontally → Failure: Global order needs one serial decision point,

while partition parallelism removes that order → Fix: Scope order to business key and partition and state the hot-key limit.**

• **Mistake: Acknowledge after the leader's local disk write → Failure: Losing the leader's availability zone can remove

the only durable copy → Fix: Acknowledge after a cross-zone commit majority and elect only a replica with the committed prefix.**

• **Mistake: Commit the offset immediately upon fetch → Failure: A crash after that commit permanently skips the business

result → Fix: Commit the idempotent business result first, then the next offset, and accept controlled replay.**

• **Mistake: Equate broker exactly-once with exactly-once external effects → Failure: The external system does not join

the broker transaction, so an acknowledgment loss still leaves a duplicate window → Fix: Use a business idempotency key, unique constraint, version condition, or reconciliation.**

• **Mistake: Immediately dead-letter every failed message → Failure: Later records for the same key can pass it and break

state order → Fix: Let the topic contract choose a paused key, paused partition, or version-based convergence.**

• **Mistake: Write a 100 MB payload directly into the broker log → Failure: A few records monopolize replication,

buffers, and fetch batches → Fix: Store the body in object storage and log its reference, size, and checksum.**

• **Mistake: Plan quotas and capacity only by message count → Failure: A 1 KB record and a 100 MB record have radically

different network, disk, and memory costs → Fix: Meter count, bytes, in-flight batches, and object concurrency.**

• **Mistake: Add consumers to eliminate every lag → Failure: One group member owns a partition at a time, and a hot key

remains limited by the serial partition path → Fix: Inspect partition and key distribution before adding partitions, splitting the business key, or throttling.**

• **Mistake: Monitor only broker uptime → Failure: A live cluster can still have lagging replicas, exhausted disks,

stale offsets, and growing dead letters → Fix: Monitor segmented latency, committed prefixes, oldest-message age, catch-up time, and an end-to-end canary.**

Follow-up Questions and Responses

Follow-up 1: How would you provide zero cross-region data loss while keeping publish p99 under 50 milliseconds?

Synchronous cross-region acknowledgment adds wide-area round-trip time to the publish path. Whether 50 milliseconds is possible depends on region distance and tail network latency. The business must rank zero RPO against local latency. When zero RPO wins, a write waits for a remote commit majority and the latency SLO must be reset. When latency wins, replication is synchronous within the region and asynchronous across regions, with an explicit risk to the unreplicated tail. Active-active also needs one owner for each key or a conflict rule; a single primary region per topic or key range usually preserves order more clearly.

Follow-up 2: One tenant has a single business key at 200,000 messages per second. Why do 128 partitions not help?

The same key must remain on one partition to preserve order, so it is still limited by the benchmarked single-partition rate of about 40,000 messages per second. The choices are to optimize the serial path, throttle that tenant, or redefine independent order domains, such as sub-entity keys that do not affect one another. If the business requires total order for that key, the service must reject a promise above serial capacity. Randomly scattering the key merely exchanges a capacity failure for an ordering failure.

Follow-up 3: A consumer charged a payment and then crashed before committing its offset. How do you prevent a second charge?

Use message_id or a business operation ID as the payment idempotency key. If the payment provider supports an idempotent API, replay uses the same key and queries the original result. If only the local database is controlled, commit business state and a processed-message unique record in one transaction, then use an outbox for the external step. When the external system has neither idempotency nor status lookup, record an UNKNOWN state, reconcile, and compensate manually. Committing the offset early only hides the uncertainty by accepting loss.

Follow-up 4: How do you redrive a poison message safely?

Repair the consumer or data first, freeze the redrive scope, and preserve the original message ID, source offset, first-seen time, and attempt history. Validate the new version with shadow consumption, then replay at a per-tenant and per-partition rate while keeping sink idempotency active. A strictly ordered topic also pauses later records for the key and resumes from the failed offset in order. If the contract permits reordering, the sink rejects old business versions. Redrive must not mint a new ID to bypass deduplication or flood the main topic during its normal traffic peak.

Follow-up 5: What changes first when retention grows from 24 hours to 30 days?

At steady state, 30 days is about 86.4 × 30 = 2.592 PB logically. Keeping three full local replicas has a lower bound near 7.776 PB, making cost and recovery time dominant. Keep active and recent segments on brokers, and upload closed, verified segments to object storage. Metadata records the object location and checksum; historical fetches use a cache or read proxy. Deletion, replay, compaction, and object lifecycle must share one retention state machine so a local segment is never deleted before its remote object is readable.