Design a Distributed Message Queue like Kafka

"Design Kafka" (or "design a distributed message queue") is a favourite infrastructure question at product companies because nearly every modern backend depends on one: Flipkart order events, Swiggy delivery updates, Razorpay payment webhooks, Zerodha market data, Hotstar analytics. Candidates who have only used Kafka through a client library often know the words (partition, offset, ISR) without knowing why each exists.

This guide derives the design from requirements, the way you should in an interview, and backs each mechanism with numbers, a dry run or code. If you want to compare brokers before going deep, read Kafka vs RabbitMQ vs SQS; for rapid-fire practice afterwards, see Kafka interview questions.

Step 1: Requirements

Functional

• Producers publish messages to named topics; consumers subscribe and receive them.
• Multiple independent consumers (fraud, analytics, notifications) can each read every message.
• Within a consumer application, work is shared across instances.
• Replay: a consumer can re-read from an earlier point after a bug fix.
• Ordering for related messages (all events of one order stay in order).
• Configurable retention by time and size; key-based compaction for "latest value per key" topics.

Non-functional

• High throughput: millions of messages per second across the cluster.
• Low latency: end-to-end in milliseconds to tens of milliseconds.
• Durability: acknowledged messages survive broker failures.
• Horizontal scalability: add brokers to add capacity.
• High availability: automatic failover of partition leadership.

Step 2: Capacity Estimates

Disk capacity, not CPU, usually decides broker count for long retention. That is why tiered storage (keeping recent segments on local disk and older ones in object storage, available in newer Kafka versions via KIP-405) is such a big deal: retention stops dictating cluster size. Practise these numbers with back-of-the-envelope estimation.

Step 3: The Key Decision: Queue or Log?

A classic queue (RabbitMQ, SQS) tracks each message's state: delivered, acknowledged, deleted. That works well for task distribution but makes "five teams each read everything" and "replay yesterday" awkward and expensive, because the broker does per-message bookkeeping for every consumer.

A log flips it: the broker just appends messages to a file and never tracks per-message consumption. Each consumer remembers a single number, its offset (position) in each partition. Reading means "give me messages from offset N". Consequences:

• Any number of consumer groups at no extra broker state.
• Replay = reset an offset.
• Deletion is by retention policy, removing old files wholesale, not per message.
• Per-message acknowledgment, delayed redelivery and priority are not native; they need patterns on top (retry topics, dead-letter topics).

Stating this choice up front, and its trade-off, frames the rest of the interview. It is also why a log is a natural foundation for patterns like event sourcing and CQRS.

Step 4: API Design

// Producer
send(topic, key, value, headers) -> Future<{partition, offset}>
  config: acks = 0 | 1 | all, linger.ms, batch.size, compression.type,
          enable.idempotence = true

// Consumer
subscribe(topics, group_id)          // partitions assigned by the group protocol
poll(timeout) -> [ {topic, partition, offset, key, value, timestamp}, ... ]
commit({partition: next_offset})     // sync or async
seek(partition, offset)              // replay or skip

// Admin
create_topic(name, partitions, replication_factor, configs)
// broker wire protocol: Produce, Fetch, Metadata, OffsetCommit, JoinGroup, ...

Note the consumer pulls. Pull lets each consumer go at its own pace, batch naturally, and rewind; with push, the broker would need per-consumer flow control. Long polling (a fetch waits up to fetch.max.wait.ms for data) avoids busy loops when the topic is quiet.

Step 5: High-Level Architecture

A topic with 3 partitions and replication factor 3: each partition has one leader and two followers on different brokers, and leaders are spread so every broker serves some traffic. Producers and consumers talk to partition leaders (consumers can optionally fetch from a nearby follower to save cross-zone traffic). The controller quorum holds cluster metadata and orchestrates failover.

Deep Dive 1: Partitions, Segments and Indexes

On disk, each partition is a directory of segments. A segment is a set of files named by the offset of its first message (its base offset):

orders-0/
  00000000000000000000.log        # message batches, append-only
  00000000000000000000.index      # sparse: relative offset -> byte position
  00000000000000000000.timeindex  # sparse: timestamp -> offset
  00000000000000368769.log        # next segment, base offset 368769
  00000000000000368769.index
  00000000000000368769.timeindex   # the newest segment is the "active" one

Only the active segment receives writes. When it reaches segment.bytes (1 GB by default) or segment.ms age, it is closed and becomes immutable. Retention then works on whole segments: deleting old data is just deleting files, which is O(1) and causes no fragmentation.

The index is sparse: an entry roughly every 4 KB of log data (index.interval.bytes). It is small enough to memory-map. To read offset X:

1. Binary search the list of segment base offsets for the largest base ≤ X.
2. Binary search that segment's index for the largest entry ≤ X, giving a byte position.
3. Scan forward in the log file from that position, at most a few KB, to the exact message.

A mini partition log you can run

import bisect


class Segment:
    def __init__(self, base_offset: int, index_interval: int):
        self.base_offset = base_offset
        self.data = bytearray()          # stands in for the .log file
        self.index = []                  # sparse (relative_offset, byte_position) = .index file
        self.count = 0
        self.bytes_since_index = index_interval   # force an index entry for the first record
        self.index_interval = index_interval

    def append(self, value: bytes) -> int:
        offset = self.base_offset + self.count
        if self.bytes_since_index >= self.index_interval:
            self.index.append((self.count, len(self.data)))
            self.bytes_since_index = 0
        record = len(value).to_bytes(4, "big") + value     # [length][payload]
        self.data += record
        self.bytes_since_index += len(record)
        self.count += 1
        return offset

    def read(self, offset: int) -> bytes:
        rel = offset - self.base_offset
        i = bisect.bisect_right(self.index, (rel, float("inf"))) - 1   # last entry <= rel
        cur_rel, pos = self.index[i]
        while True:                                     # short linear scan from there
            size = int.from_bytes(self.data[pos:pos + 4], "big")
            if cur_rel == rel:
                return bytes(self.data[pos + 4:pos + 4 + size])
            pos += 4 + size
            cur_rel += 1


class PartitionLog:
    def __init__(self, segment_bytes: int = 64, index_interval: int = 16):
        self.segment_bytes = segment_bytes
        self.index_interval = index_interval
        self.segments = [Segment(0, index_interval)]

    def append(self, value: bytes) -> int:
        active = self.segments[-1]
        if len(active.data) >= self.segment_bytes:      # roll a new segment
            active = Segment(active.base_offset + active.count, self.index_interval)
            self.segments.append(active)
        return active.append(value)

    def read(self, offset: int) -> bytes:
        bases = [s.base_offset for s in self.segments]
        seg = self.segments[bisect.bisect_right(bases, offset) - 1]   # last base <= offset
        if offset >= seg.base_offset + seg.count:
            raise IndexError("offset beyond log end")
        return seg.read(offset)


log = PartitionLog()
for i in range(20):
    log.append(f"order-{i}".encode())
print([s.base_offset for s in log.segments])   # [0, 6, 12, 18]
print(log.segments[1].index)                    # [(0, 0), (2, 22), (4, 44)]
print(log.read(13))                             # b'order-13'

import java.io.ByteArrayOutputStream;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.*;

public class PartitionLog {
    static final class Segment {
        final long baseOffset;
        final ByteArrayOutputStream data = new ByteArrayOutputStream();
        final TreeMap<Integer, Integer> index = new TreeMap<>();   // relOffset -> position
        int count = 0, bytesSinceIndex;
        final int indexInterval;

        Segment(long baseOffset, int indexInterval) {
            this.baseOffset = baseOffset;
            this.indexInterval = indexInterval;
            this.bytesSinceIndex = indexInterval;      // index the first record
        }

        long append(byte[] value) {
            if (bytesSinceIndex >= indexInterval) { index.put(count, data.size()); bytesSinceIndex = 0; }
            data.writeBytes(ByteBuffer.allocate(4).putInt(value.length).array());
            data.writeBytes(value);
            bytesSinceIndex += 4 + value.length;
            return baseOffset + count++;
        }

        byte[] read(long offset) {
            int rel = (int) (offset - baseOffset);
            Map.Entry<Integer, Integer> e = index.floorEntry(rel);   // largest entry <= rel
            ByteBuffer buf = ByteBuffer.wrap(data.toByteArray());
            int cur = e.getKey();
            buf.position(e.getValue());
            while (true) {
                int size = buf.getInt();
                if (cur == rel) { byte[] out = new byte[size]; buf.get(out); return out; }
                buf.position(buf.position() + size);
                cur++;
            }
        }
    }

    private final TreeMap<Long, Segment> segments = new TreeMap<>();
    private final int segmentBytes, indexInterval;

    PartitionLog(int segmentBytes, int indexInterval) {
        this.segmentBytes = segmentBytes;
        this.indexInterval = indexInterval;
        segments.put(0L, new Segment(0, indexInterval));
    }

    long append(byte[] value) {
        Segment active = segments.lastEntry().getValue();
        if (active.data.size() >= segmentBytes) {
            active = new Segment(active.baseOffset + active.count, indexInterval);
            segments.put(active.baseOffset, active);
        }
        return active.append(value);
    }

    byte[] read(long offset) {
        Segment s = segments.floorEntry(offset).getValue();          // largest base <= offset
        if (offset >= s.baseOffset + s.count) throw new NoSuchElementException("beyond log end");
        return s.read(offset);
    }

    public static void main(String[] args) {
        PartitionLog log = new PartitionLog(64, 16);
        for (int i = 0; i < 20; i++) log.append(("order-" + i).getBytes(StandardCharsets.UTF_8));
        System.out.println(log.segments.keySet());                                  // [0, 6, 12, 18]
        System.out.println(new String(log.read(13), StandardCharsets.UTF_8));        // order-13
    }
}

Dry run: reading offset 13

Each record is a 4-byte length plus its payload; "order-6" is 11 bytes on disk, "order-10" is 12. Segments roll once they hold 64 bytes, giving base offsets [0, 6, 12, 18]. To read 13: the largest base ≤ 13 is 12, so relative offset = 1. Segment 12's sparse index holds (0 → byte 0), (2 → byte 24)…; the largest entry ≤ 1 is (0 → 0). Scan: skip record 0 (12 bytes), land on record 1, return "order-13". In real Kafka the same three steps run with a 1 GB segment and an index entry every 4 KB, so the scan is never more than about 4 KB.

Deep Dive 2: Why a Disk-Based Log Is So Fast

• Sequential writes: appending avoids random seeks entirely.
• Page cache instead of an in-process cache: Kafka writes to the file system and lets the OS cache it. Consumers that keep up read freshly written data straight from memory. A JVM-heap cache would double memory use and suffer GC pauses.
• Batching and compression: producers accumulate messages per partition (linger.ms, batch.size) and compress the batch (lz4, zstd, snappy, gzip). Brokers store the compressed batch as-is, and consumers decompress, so the broker spends no CPU on it in the common case.
• Zero-copy: for plaintext connections the broker uses sendfile to move data from page cache to the network socket without copying through user space. (TLS requires encryption in user space, which reduces this benefit.)
• Durability via replication, not fsync: by default Kafka does not fsync every write; it relies on replicas on other machines for durability and lets the OS flush in the background.

Deep Dive 3: Producers, Partitioning and Ordering

The producer picks a partition for each message:

• With a key: partition = murmur2(key) mod numPartitions (the Java client's default). All messages with key order-123 land in the same partition, in order.
• Without a key: the "sticky" partitioner fills a batch for one partition, then switches, which gives larger batches than strict round-robin.

Ordering is guaranteed only within a partition. That is the single most important fact to state. If you need all events of one order in sequence (created → paid → shipped), key by order ID. If you need global order across all messages, you need one partition, which caps throughput at what one broker and one consumer can handle.

Retries without duplicates or reordering

If a produce request times out after the broker wrote it, a retry creates a duplicate; with several requests in flight, a retry can also reorder. The idempotent producer fixes both: the broker assigns each producer an ID, the producer numbers each batch per partition with a sequence number, and the broker rejects duplicates and out-of-sequence batches. It is on by default in recent Java clients.

Deep Dive 4: Consumer Groups, Offsets and Rebalancing

Consumers with the same group.id share work: each partition is assigned to exactly one consumer in the group. A group's maximum parallelism is therefore the partition count; a 7th consumer on a 6-partition topic sits idle. Different groups are completely independent.

Where offsets live

Consumers commit offsets to an internal, compacted topic (__consumer_offsets), keyed by (group, topic, partition). Each group is managed by a group coordinator broker (chosen by hashing the group ID onto a partition of that internal topic). Committing the offset of the next message to read is the convention.

Rebalancing

When a consumer joins, leaves or misses heartbeats (session.timeout.ms) or takes too long between polls (max.poll.interval.ms), the group rebalances. Classic "eager" rebalancing revokes all partitions from everyone, a stop-the-world pause. Improvements to mention:

• Cooperative incremental rebalancing: only partitions that move are revoked.
• Static membership (group.instance.id): a consumer restarting within the session timeout gets its partitions back without a rebalance; great for rolling deploys.
• The newer consumer group protocol (KIP-848) moves assignment logic to the broker-side coordinator to make rebalances faster and less disruptive.

Deep Dive 5: Replication, ISR and the High Watermark

Each partition has a leader and followers. Followers continuously fetch from the leader, like consumers. Each replica has a log end offset (LEO), the next offset it will write. The in-sync replica set (ISR) is the leader plus followers that have caught up within replica.lag.time.max.ms (30 seconds by default in recent versions).

The high watermark (HW) is the minimum LEO across the ISR: the boundary below which every in-sync replica has the data. Consumers can only read below the HW. That guarantees a consumer never reads a message that could vanish after a leader failover.

Dry run

Durability knobs

With RF = 3, acks=all and min.insync.replicas=2, the cluster tolerates one broker failure with no data loss and continued writes. This is the same quorum-style reasoning as in leaderless stores, but with a single leader per partition; compare with Dynamo-style quorums. Place replicas in different racks or availability zones (rack-aware assignment) so a zone outage cannot remove the whole ISR.

Deep Dive 6: Delivery Semantics

Most real systems use at-least-once plus idempotent consumers (dedupe on a message ID, or upsert by key). Kafka transactions shine in stream processing (read → transform → write) such as the pipeline in ad click aggregation. A consumer that sends an SMS or charges a card cannot be made exactly-once by Kafka; it needs its own idempotency key, which is why payment systems store processed event IDs.

Deep Dive 7: Retention, Compaction and Tiered Storage

• Delete policy (default): remove closed segments older than retention.ms (7 days by default) or beyond retention.bytes per partition. Consumers that fall further behind than retention lose data, so alert on consumer lag.
• Compact policy: keep at least the latest message per key. A background cleaner rewrites old segments dropping superseded values; a message with a null value (a tombstone) deletes the key after delete.retention.ms. Ideal for changelogs and "current state" topics: user profiles, configuration, __consumer_offsets itself.
• Tiered storage: older closed segments move to object storage while recent ones stay on local disk. Brokers become lighter, rebalancing partitions moves less data, and long retention gets cheap.

The log-shipping use case, buffering application logs for days so indexers can catch up, is a classic example of choosing retention deliberately; see distributed logging system design.

Deep Dive 8: The Controller and KRaft

Someone must decide which broker leads each partition, track the ISR, create topics and react to broker failures. That is the controller. Older Kafka stored metadata in ZooKeeper with one broker acting as controller; at large partition counts, failover meant reloading all metadata from ZooKeeper, which was slow.

KRaft replaces ZooKeeper with a small quorum of controller nodes running Raft. Metadata changes are records in a replicated metadata log; brokers follow that log and keep a local copy. Failover elects a new active controller that already has the metadata in memory, which makes recovery much faster and lets clusters support far more partitions. Kafka 4.0 removed ZooKeeper mode entirely.

Broker failure, step by step

1. The broker stops heartbeating to the controller quorum and is fenced.
2. For each partition it led, the controller picks a new leader from the remaining ISR and bumps the leader epoch.
3. The change is committed to the metadata log; brokers and clients learn the new leaders (clients get "not leader" errors and refresh metadata).
4. Producers retry to the new leader; idempotence prevents duplicates.
5. When the broker returns, it truncates any divergent tail using leader epochs, catches up and rejoins the ISR.

What Interviewers Look For

Common mistakes

• Claiming Kafka guarantees global ordering.
• Deleting messages when consumers acknowledge them, which kills replay and multi-group reads.
• Storing per-message delivery state per consumer on the broker.
• Saying acks=all alone prevents loss, without min.insync.replicas (with an ISR of just the leader, acks=all equals acks=1).
• Planning more consumers than partitions and expecting more parallelism.
• Ignoring consumer lag versus retention: a slow consumer can silently lose data when segments expire.

Follow-ups to prepare

• "How do you implement delayed retries?" Retry topics with increasing delays (retry-1m, retry-10m) and a dead-letter topic, since the log has no per-message timers.
• "How do you handle a hot partition?" Better keys (add a sub-key), more partitions for unkeyed data, or split a hot entity's stream deliberately.
• "Multi-region?" Separate clusters per region with asynchronous mirroring (MirrorMaker 2 or similar), accepting that offsets differ between clusters and that cross-region failover can lose the last few seconds of data.

Frequently Asked Questions

Pranjal Jain

Ex-Microsoft SDE · IIT Kanpur · Founder of Prepflix. Helps engineers crack startup and product company interviews across India.