Design a Pub-Sub Message Queue

Description

A publish/subscribe queue lets services react to events - "an order was placed" - without the publisher knowing who listens, and without a slow or crashed listener losing messages. In design rounds it's usually a small Kafka: an append-only log per topic, consumer groups each with their own position (offset), commits after processing, and partitions for ordering and scale. This skill builds that model and works through delivery guarantees and ordering.

"Services should react when an order is placed" often starts as the Observer pattern:

List<OrderListener> listeners = new ArrayList<>();

void placeOrder(Order order) {
    save(order);
    for (OrderListener l : listeners) l.onOrderPlaced(order);   // email, analytics, invoicing...
}

listeners: list[OrderListener] = []

def place_order(order: Order) -> None:
    save(order)
    for listener in listeners:
        listener.on_order_placed(order)   # email, analytics, invoicing...

It works until the email service is slow - then checkout is slow, because the listeners run on the caller's thread. If a listener throws, the loop stops and later listeners never hear about the order. A service that's restarting misses every order placed meanwhile, and nothing can replay history for a new service or after a bug fix. Moving to a queue that deletes each message once read fixes the timing, but now only one of the services gets each order.

We'll go from simple to more technical:

1. The model: topics as append-only logs, consumer groups with committed offsets.
2. Publishing, polling and committing.
3. Delivery guarantees: at most once, at least once, effectively once.
4. Ordering and scale with partitions, and the bugs that lose or duplicate messages.

The approach

1. Clarify. Should every interested service get every message (fan-out), or should several workers share one stream (competing consumers) - or both? Does order matter, and per what (per user, per order)? What must happen if a consumer crashes mid-message? Retention and replay? Pull or push?
2. Model. A Message(topic, offset, payload). Per topic, an append-only log. Per topic and consumer group, a committed offset. A Broker that owns them.
3. API. publish(topic, payload) returns the offset; poll(topic, group, max) returns the next messages from the group's committed offset; commit(topic, group, nextOffset); lag(topic, group).
4. Core. Reading never deletes. Each group moves forward only by committing, so an uncommitted message is delivered again. Offsets are assigned under the same lock as the append, and commits only move forward.

The pieces in code

public synchronized long publish(String topic, String payload) {
    List<Message> log = log(topic);
    Message message = new Message(topic, log.size(), payload);   // the offset and the append happen together
    log.add(message);
    return message.offset();
}

public synchronized List<Message> poll(String topic, String group, int max) {
    List<Message> log = log(topic);
    int from = (int) committedOffset(topic, group);              // a new group starts at 0
    int to = (int) Math.min(log.size(), (long) from + max);
    return List.copyOf(log.subList(from, to));                   // a copy, never a view of the log
}

public synchronized void commit(String topic, String group, long nextOffset) {
    // ... validate 0 <= nextOffset <= log size ...
    committed.get(topic).merge(group, nextOffset, Math::max);    // commits never move backwards
}

def publish(self, topic: str, payload: str) -> int:
    with self._lock:
        log = self._log(topic)
        message = Message(topic, len(log), payload)          # the offset and the append happen together
        log.append(message)
        return message.offset

def poll(self, topic: str, group: str, max_count: int) -> list[Message]:
    with self._lock:
        log = self._log(topic)
        start = self._committed_offset(topic, group)          # a new group starts at 0
        return log[start:start + max_count]                   # a slice is already a copy

def commit(self, topic: str, group: str, next_offset: int) -> None:
    with self._lock:
        ...  # validate 0 <= next_offset <= len(log)
        committed = self._committed[topic]
        committed[group] = max(committed.get(group, 0), next_offset)   # commits never move backwards

• Offset and append together. Reading log.size() and appending in two unguarded steps lets two publishers get the same offset, or one message overwrite another.
• poll doesn't move the group. Only commit does, so a consumer that crashes before committing gets the same messages again - at-least-once delivery.
• A copy, not a subList view. A view would let a consumer change the broker's log, and would break with ConcurrentModificationException as soon as the log grows. (A Python slice is always a copy, so log[start:end] gets this right for free.)
• merge(..., Math::max) ignores a late, lower commit - one that arrives after a later one - which would otherwise rewind the group and flood it with redeliveries.

Delivery guarantees

When the consumer commits, relative to processing, decides what a crash costs:

At-least-once is the usual choice, which makes idempotent consumers mandatory: remember the ids (or offsets) of messages already applied, or make the effect naturally repeatable ("set status to PAID" rather than "add 10 to balance"). A poison message - one that fails every time - would block its group forever under at-least-once, so after a few attempts it's moved to a dead-letter topic for a human to look at.

Ordering and scale

A single log is totally ordered, but only one consumer per group can read it in order. Real brokers split a topic into partitions, each its own log, and send a message to a partition by hashing its key (an order id, a user id). Order then holds per key - all events for order 42 stay in sequence - and a group scales out by giving each consumer some of the partitions. Global order across partitions is gone, which is usually fine once you pick the key well.

Where you'll see it

This project is a small Kafka: topics made of partitioned logs, consumer groups with committed offsets, retention by time or size, and replay by resetting offsets. Queue systems such as RabbitMQ and Amazon SQS take the other design: a message goes to one consumer and is deleted when acknowledged (SQS hides a received message for a visibility timeout and redelivers it if it isn't deleted in time; after too many receives a redrive policy moves it to a dead-letter queue). Push-based brokers must handle backpressure - a slow consumer can't be sent messages faster than it processes them, so they cap in-flight messages per consumer; pull-based consumers set their own pace.

Which tool for which job?

Common mistakes

Check yourself

Try to answer before opening each one - recalling beats re-reading.

1. The email service and the analytics service both need every order event; the email service runs three instances. How many consumer groups, and why?

2. A consumer commits the offset, then charges the customer, and crashes between the two. What happens to that payment - and what if it charges first, then commits?

3. Why must publish assign the offset and append the message under one lock?

Key takeaways

• A broker decouples producers from consumers in time: append-only logs, and consumers that pull at their own pace.
• Each consumer group has its own committed offset: fan-out across groups, shared work within one, replay by moving the offset.
• poll doesn't advance; commit does, and never backwards - that's at-least-once, so consumers must be idempotent.
• Order holds per log or partition; choose the partition key so the events that must stay in order share one.
• Assign offsets inside the append's critical section, return copies, and watch lag against retention.