Distributed Message Queue

Design a distributed message queue with event-streaming features: long data retention (2 weeks), repeated consumption, message ordering.

Benefits of message queues: decoupling, improved scalability (scale producers/consumers independently), increased availability, better performance (async communication).

Note: traditional message queues (RabbitMQ, ActiveMQ) differ from event streaming platforms (Kafka, Pulsar), but features are converging (RabbitMQ added streams; Pulsar works as both). This design targets the richer feature set.

Step 1 of 4: Understand the problem and establish design scope #

Goal: lock in requirements before designing.

Functional requirements #

Must have:

1. Producers send messages; consumers consume them
2. Messages can be consumed repeatedly or only once
3. Historical data truncated (retention: 2 weeks)
4. Message size: KB range, text only
5. Messages consumed in production order
6. Delivery semantics: at-least-once, at-most-once, exactly-once (configurable)
7. High throughput for log aggregation; low latency for traditional queue use cases

Non-functional:

1. Configurable throughput vs latency
2. Scalable and distributed; handles traffic surges
3. Persistent and durable: data persisted on disk, replicated across nodes

Traditional message queue adjustments: traditional queues have no strong retention, no ordering guarantees, smaller on-disk overflow. These simplifications reduce design complexity (noted where applicable).

Step 2 of 4: Propose high-level design and get buy-in #

Goal: agree on the core abstractions: topics, partitions, brokers, consumer groups.

Figure 2: Key components

Messaging models #

1. Point-to-point: one message → one consumer. Traditional model. Message removed after ack
2. Publish-subscribe: one message → all subscribers of a topic

This design supports both. Pub/sub via topics; point-to-point via consumer groups (single consumer in group per partition).

Topics, partitions, and brokers #

1. Topic: logical category. Partition: shard of a topic. Messages distributed across partitions
2. Partition = FIFO queue. Position = offset
3. Message key → hash(key) % numPartitions determines partition. No key → random
4. Broker: server holding partitions. Partitions distributed across brokers for scalability
5. Consumer: subscribes to a topic, pulls from one or more partitions

Consumer group #

Set of consumers working together. Each group maintains its own consuming offsets. One partition → one consumer per group (constraint for ordering). More partitions than consumers → scale by adding consumers.

If all consumers are in one group → point-to-point model (each partition consumed by exactly one consumer).

High-level architecture #

Figure 8: High-level design

1. Clients: producers push to topics; consumer groups subscribe
2. Broker: holds partitions
3. Storage: data storage (messages in partitions), state storage (consumer offsets), metadata storage (topic config, partitions, retention, replica distribution)
4. Coordination service: service discovery (alive brokers), leader election (active controller assigns partitions). Uses Zookeeper or etcd

Step 3 of 4: Design deep dive #

Goal: three key design choices for high throughput + data retention:

1. On-disk data structure optimized for sequential access (rotational disk)
2. Message data structure allows zero-copy from producer → queue → consumer
3. Batching everywhere: producers, queue, consumers

Data storage #

Traffic pattern: write-heavy, read-heavy. No updates/deletes. Predominantly sequential access.

1. Option 1: database. Not ideal. Hard to design for both write-heavy + read-heavy at scale
2. Option 2: write-ahead log (WAL). Recommended. Append-only log file. Sequential read/write. Rotational disks in RAID → several hundred MB/sec. OS aggressively caches disk data in main memory

Segments: active segment (appends) → when full → becomes inactive (read-only). Old segments truncated after the retention period.

Message data structure #

1. Key: determines partition via hash(key) % numPartitions. Not unique. Not equivalent to partition number
2. Value: payload (plain text or compressed binary)
3. CRC: integrity check

No data copying through the pipeline: producers, queue, and consumers agree on the format.

Batching #

Critical for performance:

1. OS groups messages in a single network request (amortizes round trips)
2. Brokers write in large chunks → larger sequential writes, larger contiguous disk cache

Tradeoff: large batch → higher throughput, higher latency. Configurable per use case.

Producer flow #

Option 1 (routing layer): extra network hop, no batching → rejected.

Option 2 (routing in the producer client library): improved. Buffer + routing in the producer. Benefits: fewer hops, producer controls partition logic, batching in memory → large batches → higher throughput.

Batch size tradeoff: large = high throughput + high latency; small = low latency + low throughput.

Consumer flow (pull model) #

Push vs Pull:

Pull model chosen. Long polling: wait a specified time for new messages when the queue is empty.

Consumer flow:

1. New consumer joins a group, finds the coordinator by hashing the group name
2. Coordinator assigns partitions (round-robin, range, etc.)
3. Consumer fetches from the last committed offset (state storage)
4. Consumer processes, commits the offset. Commit timing affects delivery semantics

Consumer rebalancing: triggered by consumer join/leave/crash or partition change. The coordinator manages heartbeats, elects a group leader; the leader generates a partition dispatch plan → coordinator broadcasts.

State storage #

Stores: partition↔consumer mapping, last consumed offset per consumer group per partition.

Access pattern: frequent read/write (low volume), rarely deleted, random access, consistency important → Zookeeper (KV store) is a good fit (Kafka later moved offset storage to brokers).

Metadata storage #

Topic config: number of partitions, retention period, replica distribution. Low volume, high consistency → Zookeeper.

Zookeeper consolidation #

Zookeeper handles: metadata storage, state storage (consumer offsets), broker leader election. The broker now only maintains message data storage.

Replication #

Each partition: 3 replicas across different brokers. Leader handles writes; followers pull from the leader. Producer acks after "enough" replicas sync.

1. Replica distribution plan: generated by the active controller (elected broker). Persisted in metadata storage
2. In-sync replicas (ISR): replicas caught up with the leader (within configurable lag). Leader always in ISR

Acknowledgment settings:

Consumers read from leader replicas (simplicity, limited connections per partition). If a topic is hot → scale partitions + consumers. Alternative: read from the closest ISR for cross-DC scenarios.

Scalability #

Producers: easy: add/remove instances.

Consumers: consumer groups are isolated → easy to add/remove groups. Rebalancing handles group-internal changes.

Brokers (failure recovery):

When a broker crashes → the controller detects it and generates a new replica distribution plan. New replicas catch up as followers.

Fault-tolerance considerations:

1. Min ISR count: balances latency vs safety
2. Replicas on different nodes (never the same broker)
3. Cross-DC replicas → safer but higher latency/cost. Data mirroring as a workaround

Adding brokers: the controller temporarily allows extra replicas → new broker catches up → remove redundant replicas. No data loss.

Partitions:

1. Increasing: new messages go to all partitions; old messages stay
2. Decreasing: the decommissioned partition stops receiving new messages but remains readable until retention expires. No immediate data migration

Data delivery semantics #

At-most-once:

1. Producer: async, ack=0, no retry
2. Consumer: commit offset BEFORE processing. Crash after commit → message lost
3. Use: metrics/logging (small data loss acceptable)

At-least-once:

1. Producer: ack=1 or ack=all, retry on failure/timeout
2. Consumer: commit offset AFTER processing. Crash after processing but before commit → re-consume (duplicate)
3. Use: most cases; deduplication handled by a unique key on the consumer side

Exactly-once:

1. Most difficult. High performance/complexity cost
2. Use: financial (payment, trading, accounting) where duplication is unacceptable

Step 4 of 4: Wrap up #

Advanced features:

Message filtering #

Problem: the consumer needs only a subset of a topic's messages. Naive: consumer fetches all, filters client-side (wasteful). Better: broker-side filtering via message tags (metadata, not payload: avoids decryption/deserialization overhead). Tags support multi-dimensional filtering. Complex logic (formulae) → too heavyweight for a message queue.

Delayed & scheduled messages #

Delayed: message held in temporary storage, delivered to the topic after a specified delay. Example: order payment check after 30 min.

Timing implementations:

1. Dedicated delay queues with predefined levels (RocketMQ: 1s, 5s, 10s, 30s, 1m, 2m, 3m, 4m, 6m, 8m, 9m, 10m, 20m, 30m, 1h, 2h)
2. Hierarchical time wheel

Scheduled: similar design; message delivered at a specific scheduled time.

Additional talking points:

1. Protocol: AMQP, Kafka protocol: cover all activities (produce, consume, heartbeat), transport large data, data integrity
2. Retry consumption: failed messages → dedicated retry topic for later consumption
3. Historical data archive: truncated messages archived to HDFS or object storage for replay