Distributed Message Queue

Problem Statement & Requirements

Design a distributed message queue system that decouples producers from consumers, enables asynchronous processing, guarantees message delivery, and scales to handle millions of messages per second — similar to Apache Kafka, AWS SQS, or RabbitMQ.

Functional Requirements

• Producers publish messages to named topics
• Consumers subscribe to topics and receive messages in order
• Support consumer groups — multiple consumers sharing the load of a topic, each message delivered to exactly one consumer in the group
• Messages are persisted on disk, not lost on broker failure
• Support at-least-once, at-most-once, and exactly-once delivery semantics (configurable)
• Consumers can replay messages from a specific offset (time-travel)
• Support message retention by time (e.g., 7 days) or size (e.g., 1 TB per topic)
• Ordering guarantees within a partition (not across partitions)
• Dead letter queue (DLQ) for messages that fail processing after N retries

Non-Functional Requirements

• High throughput — handle millions of messages/second per cluster
• Low latency — end-to-end publish-to-consume in < 10 ms (p99) for real-time use cases
• Durability — zero message loss once acknowledged by the broker
• High availability — survive broker failures, rack failures, even AZ failures
• Horizontal scalability — add brokers to increase throughput linearly
• Fault tolerance — automatic leader election and partition rebalancing on failure

Out of Scope

• Complex message routing (fanout, topic exchanges like RabbitMQ — simplified to topic-partition model)
• Message transformation / stream processing (that's Kafka Streams / Flink territory)
• Multi-tenancy and authentication (simplify for interview)

Scale Estimations

Traffic

Storage

Bandwidth

Hardware Estimate (per broker)

This first layer defines the system boundary. The next chapters add application logic and data dependencies. After examining the request paths and tradeoffs, we will add operational safeguards and bring the complete architecture together.

API Design

Producer API

POST /api/v1/topics/{topic_name}/messages
Content-Type: application/json

Request:
{
  "key": "user-123",                    // partition key (optional — null = round-robin)
  "value": "base64-encoded-payload",     // the message body
  "headers": {                           // optional metadata
    "correlation-id": "req-abc",
    "content-type": "application/json"
  },
  "partition": 3,                        // optional — explicit partition override
  "timestamp": 1712150400000             // optional — defaults to broker time
}

Response (202 Accepted):
{
  "topic": "user-events",
  "partition": 3,
  "offset": 847291,
  "timestamp": 1712150400000
}

Consumer API

GET /api/v1/topics/{topic_name}/messages?group_id=analytics-pipeline
    &max_messages=500
    &timeout_ms=5000

Response (200 OK):
{
  "messages": [
    {
      "key": "user-123",
      "value": "base64-encoded-payload",
      "headers": {"correlation-id": "req-abc"},
      "partition": 3,
      "offset": 847291,
      "timestamp": 1712150400000
    },
    ...
  ]
}

Commit Offset (Acknowledge)

POST /api/v1/topics/{topic_name}/offsets
Content-Type: application/json

Request:
{
  "group_id": "analytics-pipeline",
  "offsets": [
    {"partition": 0, "offset": 12345},
    {"partition": 3, "offset": 847292}
  ]
}

Response: 204 No Content

Admin API

POST   /api/v1/topics                     — Create topic (name, partitions, replication_factor, retention)
DELETE /api/v1/topics/{topic_name}        — Delete topic
PUT    /api/v1/topics/{topic_name}/config — Update retention, compaction, etc.
GET    /api/v1/topics/{topic_name}/info   — Topic metadata (partitions, leaders, ISR)
POST   /api/v1/topics/{topic_name}/partitions — Add partitions (cannot decrease)

SDK-Level API (Internal Binary Protocol — High Performance)

// Producer (batch-oriented)
producer.send(topic, key, value, headers) → Future<RecordMetadata>
producer.flush()                          → blocks until all buffered messages sent

// Consumer (poll-based)
consumer.subscribe(topics, group_id)
records = consumer.poll(timeout_ms)       → batch of ConsumerRecords
consumer.commitSync(offsets)              → block until offsets durably stored
consumer.commitAsync(offsets, callback)   → fire-and-forget with callback
consumer.seek(partition, offset)          → rewind/fast-forward to specific offset

Data Model

Core Abstractions

Message Record (On-Disk Format)

Log Segment (On-Disk Storage Unit)

Partition Directory: /data/user-events-0/

  00000000000000000000.log      ← active segment (append-only)
  00000000000000000000.index    ← sparse offset → file position index
  00000000000000000000.timeindex← timestamp → offset index
  00000000000065536000.log      ← rolled segment (immutable)
  00000000000065536000.index
  00000000000065536000.timeindex

Segment rolls when:
  - Size exceeds segment.bytes (default 1 GB)
  - Time exceeds segment.ms (default 7 days)
  - Index file is full

Offset Storage

Core Design Decisions — The Storage Engine

The most critical design decision in a message queue is how messages are stored and retrieved. This determines throughput, latency, durability, and operational complexity.

Option 1: Append-Only Log on Disk (Kafka's Approach)

Why sequential I/O matters:

Sequential I/O is 100-1000× faster than random I/O. By only appending, Kafka achieves disk throughput that rivals network throughput.

Zero-copy transfer (sendfile):

Traditional path:                    Zero-copy path:
Disk → Kernel buffer                 Disk → Kernel buffer
Kernel buffer → User buffer             ↓  (sendfile syscall)
User buffer → Socket buffer          Kernel buffer → NIC buffer
Socket buffer → NIC buffer

4 copies, 4 context switches         2 copies, 2 context switches
~60% of CPU on data copies           ~0% CPU on data copies

Option 2: In-Memory Queue with WAL (RabbitMQ-style)

Option 3: Database-Backed Queue (SQS-style)

Recommendation

Use Option 1 (Append-Only Log) for a high-throughput, scalable message queue:

• Sequential I/O + zero-copy = unmatched throughput
• Persistent storage = unlimited retention and replay
• Simple broker logic = operational simplicity
• Consumer offset tracking = flexible consumption patterns

Choose the ordering boundary before choosing the partition count. Events for one business key can share a partition, but separate partitions progress independently. A consumer acknowledgement says how far processing has advanced under its contract; it does not make an external database write and an offset commit automatically atomic.

Detailed Flow Diagrams

Produce Flow

Consume Flow (Pull-Based)

Consumer Rebalance Flow

Log Compaction (Bonus Deep Dive)

Replication & Consistency

ISR (In-Sync Replicas)

Acks Configuration & Durability Tradeoffs

Recommendation: acks=all with min.insync.replicas=2 for production:

• Tolerates 1 broker failure without data loss
• Blocks produces if ISR shrinks below 2 (prevents single-replica writes)

Leader Election

Partitioning Strategy

How Messages Are Routed to Partitions

Partition Count Selection

Target throughput: 1 GB/s

Single partition throughput:
  - Producer: ~10-50 MB/s per partition
  - Consumer: ~20-100 MB/s per partition (consumer is usually faster)
  - Bottleneck is typically producer side

Calculation:
  Required partitions = Target throughput / Per-partition throughput
                      = 1 GB/s / 30 MB/s (conservative)
                      = ~34 partitions per topic (round up to 36-48)

Rule of thumb:
  - Start with max(expected_throughput / 30MB, num_consumers_in_largest_group)
  - Can always ADD partitions later (cannot remove)
  - More partitions = more parallelism BUT more open file handles,
    longer leader election, more memory for consumer offset tracking

Partition-to-Consumer Assignment Strategies

Recommendation: Cooperative Sticky Assignor — minimizes stop-the-world rebalancing.

Handling Edge Cases

Consumer Lag & Backpressure

Problem: Consumer is slower than producer → unbounded lag
  
Monitoring:
  consumer_lag = latest_offset(partition) - committed_offset(consumer_group, partition)
  
Mitigations:
  1. Scale out consumers (up to # of partitions)
  2. Increase consumer fetch size (fetch.max.bytes)
  3. Tune consumer processing (batch DB writes, parallelize)
  4. Add partitions to topic (+ proportional consumers)
  5. Alert when lag exceeds threshold (e.g., > 100K messages)

Exactly-Once Semantics (EOS)

Dead Letter Queue (DLQ)

Message Ordering Across Partitions

Problem: User actions must be ordered, but topic has 12 partitions

Solution: Use the entity ID as the partition key
  - key = "user-123" → hash("user-123") % 12 = partition 7
  - ALL events for user-123 go to partition 7
  - Within partition 7, ordering is guaranteed

Caveat: If you add partitions, hash changes → key mapping shifts
  - Mitigate: Don't add partitions to topics where key ordering matters
  - Or: use a custom partitioner with a stable mapping table

Large Messages

Tradeoffs & Design Decisions Summary

Reliability & Fault Tolerance

Single Points of Failure & Mitigations

Graceful Degradation

Scenario: Broker goes down (1 of 50)

  1. Controller detects missing heartbeat (18s timeout)
  2. For each partition where dead broker was leader:
     → Elect new leader from ISR
     → Update metadata, notify clients
  3. For each partition where dead broker was follower:
     → Remove from ISR
     → Continue with remaining ISR
  4. When broker returns:
     → Fetches missed data from current leaders
     → Rejoins ISR after catching up
     → Preferred leader election restores original topology

Impact: ~5-30s of unavailability for affected partitions
         Other partitions continue unaffected
         Zero data loss (acks=all with min.insync.replicas=2)

Data Durability Guarantees

High-Level Architecture

Component Breakdown

Key: P0-L = Partition 0 Leader, P0-F = Partition 0 Follower

Full System Architecture (Production-Grade)

Let a consumer complete a database write and crash before committing its offset. On restart, the message is delivered again. Design the deduplication or transactional boundary that makes this safe, then check what changes if the side effect is a third-party API rather than your own database.

Interview Tips

1. Start with the append-only log insight — This is the foundational design decision. Explain why sequential I/O and zero-copy make disk-based queues faster than in-memory alternatives at scale.

2. Draw the partition model early — Topics → Partitions → Segments. This shows you understand the parallelism model. Emphasize: ordering is per-partition, not per-topic.

3. Discuss the acks tradeoff in depth — acks=0 vs acks=1 vs acks=all. Connect it to ISR and min.insync.replicas. This is a nuanced durability vs latency discussion interviewers love.

4. Explain consumer groups and rebalancing — How partitions are assigned, what triggers a rebalance, why cooperative sticky is preferred over stop-the-world. Mention the "consumers ≤ partitions" constraint.

5. Address exactly-once semantics — Most candidates only know at-least-once. Explaining idempotent producers (PID + sequence number) and transactions sets you apart.

6. Mention KRaft over ZooKeeper — Shows you know the modern architecture. ZooKeeper was a bottleneck at >200K partitions. KRaft uses Raft consensus built into the brokers themselves.

7. Discuss operational concerns — Partition count selection, when to add partitions (and why you can't remove them), retention policies, monitoring consumer lag, dealing with hot partitions.

8. Claim-check pattern for large messages — Shows pragmatism. Don't just increase max.message.bytes — explain the S3 reference pattern.

9. Compare with alternatives when asked — Kafka (high throughput, log-based) vs RabbitMQ (flexible routing, push-based) vs SQS (managed, simple). Each has its sweet spot.

10. End with scale numbers — "A single Kafka cluster with 50 brokers can handle 1M+ messages/sec, store petabytes with 7-day retention, and serve hundreds of consumer groups simultaneously."