Design a distributed message queue (like Kafka)

Traditional brokers (RabbitMQ, ActiveMQ) track per-message state and delete messages after acknowledgement, which means random I/O and per-message bookkeeping. Kafka's design is a partitioned, append-only log: the broker only appends and serves ranges by offset, and consumers keep their own position. This makes several things cheap. Writes are sequential appends to segment files, which disks handle at hundreds of MB/s. Reads of recent data come from the OS page cache, and the broker uses zero-copy sendfile to move bytes from page cache to the socket without copying them into the JVM. Producers batch many records and compress the whole batch, and the broker stores the batch as-is, so compression and batching benefits carry from producer to consumer. Because messages are not deleted on consumption, any number of consumer groups can read the same data, and a group can rewind to reprocess after a bug fix. The cost: ordering exists only within a partition, there is no per-message acknowledgement or delay, and a slow message blocks the rest of its partition for that group (head-of-line blocking). Mention that SQS/RabbitMQ are better for per-message work queues with individual retries.

Each partition has one leader and followers on other brokers, usually in different AZs (replication factor 3, rack-aware placement). Followers pull from the leader like consumers. The leader tracks the in-sync replica set (ISR): followers that have caught up within replica.lag.time.max.ms (30 s by default in recent versions). The high watermark is the highest offset replicated to all ISR members, and only data below it is visible to consumers. With acks=all, the leader acknowledges a write only when all current ISR members have it, and min.insync.replicas=2 refuses writes if the ISR shrinks below 2, so an acknowledged write is always on at least 2 brokers. If the leader dies, the controller picks a new leader from the ISR, so no acknowledged data is lost. Setting unclean.leader.election.enable=false prevents an out-of-sync replica from becoming leader (which would lose data) at the cost of the partition staying offline until an ISR member returns. Leader epochs let followers truncate any uncommitted tail they had from the old leader so all replicas converge. A common interview mistake is to say 'acks=all means all 3 replicas'; it means all in-sync replicas, which could be fewer, which is why min.insync.replicas matters.

Out of the box, a consumer that commits after processing gets at-least-once: a crash between processing and commit causes reprocessing. Committing before processing gives at-most-once (messages can be skipped). On the producer side, a retry after a timeout can write the same message twice; the idempotent producer (enabled by default since Kafka 3.0) attaches a producer ID and per-partition sequence numbers so the broker drops duplicates. Kafka transactions let a producer write to several partitions and commit consumer offsets atomically, so a read-process-write pipeline within Kafka (for example Kafka Streams or Flink with the Kafka sink) is exactly-once, and consumers with isolation.level=read_committed skip aborted data. But this guarantee stops at Kafka's boundary. If the consumer writes to Postgres or calls a payment API, you get effectively-once only by making that side idempotent: store the processed offset in the same DB transaction as the result, or use a business idempotency key with a unique constraint. Interviewers listen for exactly this distinction.

Partitions are the unit of parallelism and ordering. A consumer group can have at most as many active consumers as partitions, so a topic with 12 partitions cannot use 40 consumers. Choosing the key decides ordering and balance: key by order_id to keep each order's events ordered; do not key by merchant_id if one big merchant produces 30% of traffic, because that creates a hot partition. Too many partitions also cost: more open files, more leader elections during failover, longer metadata updates and more memory on clients; KRaft raised practical limits to millions per cluster, but each partition still has overhead. Increasing the partition count later changes key-to-partition mapping and breaks per-key ordering during the transition, so pick a generous count up front (e.g. based on target throughput ÷ ~10 MB/s per partition, and maximum consumers needed). For slow processing, scale within a partition by processing different keys in parallel inside the consumer while committing offsets carefully (Confluent's parallel consumer pattern).

Kafka gives replay, many independent consumers, high throughput and per-key ordering, but no per-message ack, retries or delays, and needs partition planning. SQS/RabbitMQ give per-message acknowledgement, visibility timeouts, dead-letter queues and delays with no partition planning, but messages are gone once consumed and throughput per queue is lower. Many systems use both: Kafka for event streams, SQS for task queues.

Kafka by default does not fsync every write; it relies on replicas in different AZs, since a simultaneous power loss in three AZs is far less likely than one disk failing. Forcing fsync per batch improves single-node durability but cuts throughput significantly. For most systems, acks=all + min.insync.replicas=2 + rack-aware placement is the right trade-off.

Keeping 7 days on local disks needs many brokers mainly for storage, and rebalancing moves terabytes. Tiered storage (KIP-405: early access in Kafka 3.6, production-ready in 3.9) keeps hours locally and older segments in S3, making brokers lighter and retention cheap, at the cost of slower reads of old data and dependence on the object store.

A skewed key sends far more traffic to one partition than others, overloading one broker and one consumer. Fix the key (add a sub-key or salt for hot entities when strict ordering per entity is not needed), or give hot tenants dedicated topics.

On a Big Billion Days sale, order events can jump 10×. If consumers fall behind, lag grows but no data is lost while within retention. Autoscale consumers up to the partition count, alert on lag in time (not just messages), and make sure catch-up reads that miss page cache do not starve real-time readers (tiered storage helps isolate this).

The controller notices the missing heartbeat (a few seconds), and for each partition that broker led it elects a new leader from the ISR. Producers and consumers get NOT_LEADER errors, refresh metadata and continue. Acknowledged data is safe because it was on at least min.insync.replicas brokers.

With rack-aware placement, each partition has replicas in 3 AZs; losing one leaves 2, which still meets min.insync.replicas=2. Clusters should run under ~60% capacity so the remaining brokers can absorb leadership and traffic.

A message that always fails blocks its partition for that group. Catch the error, publish the message to a dead-letter topic with the error, commit past it, and alert. Without this, one poison message can stop order processing.