Pub-sub message broker

Design an in-memory message broker: publishers append messages to topics, and each subscriber reads them at its own pace with its own offset. A slow or failing subscriber must never hold up the others.

Design an in-memory publish-subscribe message broker for one process. Publishers send messages to named topics. Every subscriber of a topic receives every message published after it subscribed. Subscribers can be slow, can fail, and can leave while messages are being delivered.

Questions to ask first

Is this one process or a network service? Assume one process: publishers and subscribers are threads of the same program. There is no network and no disk.
What does a new subscriber receive? Assume every message published after it subscribed, in the order published. Older messages come only if the subscriber asks to replay them with a seek.
What delivery guarantee do we promise? Assume at-least-once inside the process: a message is offered again until the handler confirms it. We do not promise exactly-once.
What if one subscriber is slow or keeps failing? Assume the publisher and the other subscribers must not wait for it. The slow one falls behind on its own.
What order do we promise? Assume order per topic: each subscriber sees messages in the order they were published to that topic. There is no order across topics.
Can a subscriber leave in the middle of a delivery? Yes, and a handler may even remove itself. After unsubscribe returns, that subscriber gets no more messages.

Requirements

Create topics. Subscribe and unsubscribe handlers by name.
Publish appends a message to a topic and returns its offset: 0 for the first message, then 1, 2 and so on.
Every subscription keeps its own offset, so each subscriber reads at its own pace. A new subscriber starts at the end of the topic, and a seek can move it back.
Deliver every message to every subscriber at least once, in offset order. The offset moves forward only after the handler confirms the message.
A slow or failing subscriber must not block the publishers or the other subscribers.
Unsubscribe takes effect at once, also during a delivery and also from inside a handler.
Offer a synchronous mode with an explicit drain call, so that tests do not depend on timing.

Out of scope

A network protocol, disk storage and surviving a restart.
Consumer groups and partitions (an extension below).
Routing by message content and wildcard topics.
Exactly-once delivery.

Entities

Broker: createTopic(name), publish(topic, payload), unsubscribe(topic, name), drain()
Topic: name, an append-only log of messages; append(payload), read(offset), subscribe(name, handler)
Message: offset, payload
Subscriber: the handler interface, onMessage(topic, message) returns true when confirmed
Subscription: one subscriber on one topic: nextOffset, active; deliverPending(topic), seek(offset), cancel()
Dispatcher: decides who runs the deliveries; SynchronousDispatcher (drain on the caller's thread), LaneDispatcher (a thread per subscriber)

Classes in the model design

Broker (class)
Topic (class)
Message (class)
Subscriber (interface)
Subscription (class)
Dispatcher (interface)
SynchronousDispatcher (class)
LaneDispatcher (class)

The main use case, step by step

The app creates the topic `orders` and subscribes `billing` and `email`. Each subscription starts at the end of the log, so both have next offset 0.
A publisher calls `broker.publish("orders", "order-1")`. The broker finds the topic, which appends the message to its log and returns offset 0. The publisher does not wait for any subscriber.
Later `broker.drain()` runs. For each topic the dispatcher walks the subscriptions in the order they subscribed. Each subscription delivers its waiting messages, in offset order, before the next one starts.
`billing` receives `order-1`. Its handler returns true, so the subscription moves its offset from 0 to 1. `email` does the same with its own offset.
Now `order-2` arrives and the `billing` handler throws. The subscription catches the error and leaves its offset at 1. It stops for this drain, and `email` is served as usual.
At the next drain `billing` is offered `order-2` again. This is at-least-once delivery: the same message may reach a handler more than once, and never less.
If a handler calls `unsubscribe` while the drain runs, the subscription is marked inactive first. Its loop checks the flag before each message, so it stops at once.

Patterns used

Observer: Topic notifies its Subscription objects, which call their Subscriber handlers.. The publisher knows only the topic. New subscribers join without any change to the publisher. The offset and the log add what plain observer lacks: each observer reads at its own pace.
Strategy: Dispatcher, with SynchronousDispatcher for tests and LaneDispatcher for real use.. How deliveries run (same thread, a thread per subscriber, a pool) changes more often than what a delivery is. Behind an interface, tests stay deterministic and production gets threads.
Facade: Broker is the only class the outside code calls.. Callers never see logs, subscriptions or locks. The rules (offset moves after confirmation, unsubscribe stops delivery) live in one place and cannot be skipped.

Key code

Message, Subscriber and Subscription: the offset lives here

public record Message(int offset, String payload) {}

/** The handler. Return true to confirm the message. Return false, or throw, to get it again. */
public interface Subscriber {
    boolean onMessage(String topic, Message message);
}

/** One subscriber's place in one topic: its own offset, and a lock that keeps its deliveries in order. */
public final class Subscription {
    private final String name;
    private final Subscriber subscriber;
    private final ReentrantLock lock = new ReentrantLock();   // reentrant: a handler may unsubscribe itself
    private volatile boolean active = true;
    private int nextOffset;                                   // offset of the next message to deliver

    public Subscription(String name, Subscriber subscriber, int startOffset) {
        this.name = name;
        this.subscriber = subscriber;
        this.nextOffset = startOffset;
    }

    /** Delivers every waiting message in order. Stops at the first failure and keeps the offset there. */
    public int deliverPending(Topic topic) {
        int delivered = 0;
        lock.lock();
        try {
            while (active && nextOffset < topic.size()) {     // active is checked before every message
                Message m = topic.read(nextOffset);
                boolean confirmed;
                try { confirmed = subscriber.onMessage(topic.name(), m); }
                catch (RuntimeException e) { confirmed = false; }
                if (!confirmed) break;                        // offered again at the next drain
                nextOffset = m.offset() + 1;                  // move on only after the confirmation
                delivered++;
            }
        } finally {
            lock.unlock();
        }
        return delivered;
    }

    public void cancel() { active = false; }
    public String name() { return name; }
}

Topic and the dispatcher that runs deliveries

public final class Topic {
    private final String name;
    private final List<Message> log = new ArrayList<>();                          // the offset is the index
    private final List<Subscription> subscriptions = new CopyOnWriteArrayList<>();  // safe to walk while a handler unsubscribes

    public Topic(String name) { this.name = name; }

    public synchronized Message append(String payload) {
        Message m = new Message(log.size(), payload);
        log.add(m);
        return m;
    }

    public synchronized int size() { return log.size(); }
    public synchronized Message read(int offset) { return log.get(offset); }
    public String name() { return name; }
    public List<Subscription> subscriptions() { return subscriptions; }

    /** A new subscriber starts at the end of the log. */
    public synchronized Subscription subscribe(String subscriber, Subscriber handler) {
        Subscription s = new Subscription(subscriber, handler, log.size());
        subscriptions.add(s);
        return s;
    }

    public boolean unsubscribe(String subscriber) {
        for (Subscription s : subscriptions) {
            if (s.name().equals(subscriber)) {
                s.cancel();                      // no more messages from this moment, even in a drain that is running
                return subscriptions.remove(s);
            }
        }
        return false;
    }
}

public interface Dispatcher {
    /** Gives every subscriber of the topic its waiting messages. Returns the deliveries it finished itself. */
    int dispatch(Topic topic);
}

/** Delivery on the caller's thread, one subscriber after the other. Deterministic, so tests use it. */
public final class SynchronousDispatcher implements Dispatcher {
    @Override public int dispatch(Topic topic) {
        int delivered = 0;
        for (Subscription s : topic.subscriptions()) delivered += s.deliverPending(topic);
        return delivered;
    }
}

Broker, and one lane per subscriber for real threads

public final class Broker {
    private final Map<String, Topic> topics = new LinkedHashMap<>();
    private final Dispatcher dispatcher;

    public Broker(Dispatcher dispatcher) { this.dispatcher = dispatcher; }

    public synchronized boolean createTopic(String name) {
        return topics.putIfAbsent(name, new Topic(name)) == null;
    }

    public synchronized Topic topic(String name) { return topics.get(name); }

    public int publish(String topicName, String payload) {
        Topic t = topic(topicName);
        return t == null ? -1 : t.append(payload).offset();     // the publisher never waits for a subscriber
    }

    public boolean unsubscribe(String topicName, String subscriber) {
        Topic t = topic(topicName);
        return t != null && t.unsubscribe(subscriber);          // marks the subscription inactive, then removes it
    }

    public int drain() {
        List<Topic> all;
        synchronized (this) { all = new ArrayList<>(topics.values()); }
        int delivered = 0;
        for (Topic t : all) delivered += dispatcher.dispatch(t);
        return delivered;
    }
}

/** One single-thread lane per subscriber: a slow subscriber only delays its own lane. */
public final class LaneDispatcher implements Dispatcher {
    private final Map<Subscription, ExecutorService> lanes = new ConcurrentHashMap<>();

    @Override public int dispatch(Topic topic) {
        for (Subscription s : topic.subscriptions()) {
            lanes.computeIfAbsent(s, k -> Executors.newSingleThreadExecutor())
                 .execute(() -> s.deliverPending(topic));       // runs in the lane, in order
        }
        return 0;                                               // nothing is finished yet: lanes work on their own
    }

    public void shutdown() { lanes.values().forEach(ExecutorService::shutdown); }
}

Deep dives

A slow subscriber must not block the others

Weak: publish calls every handler in a loop on the publisher's thread. One handler that takes 5 seconds delays the publisher and every subscriber after it. One handler that throws stops the loop, and the rest never see the message.

Good: A queue per subscriber and one thread per subscriber. The publisher only appends. Each thread reads its own queue, so a slow handler delays only itself. It is easy to explain and the isolation is strict. But 10,000 subscribers mean 10,000 threads, and each one costs memory.

Great: Keep the log and one offset per subscriber, which is already a queue that costs nothing to hold. Let a small shared pool run the deliveries, with one rule: at most one task per subscription at a time. That keeps each subscriber's order. A slow subscriber holds at most one pool thread, and a handler timeout frees even that one.

Trade-off: A thread per subscriber is simplest and fully isolated. A pool scales better but needs the one-task rule and a timeout. In both, a subscriber that falls far behind needs a policy: block the publisher, drop its oldest messages, or disconnect it. Say which one you pick and why.

What delivery guarantee do we give?

Weak: Move the offset forward first, then call the handler. If the handler fails or the process stops in between, the message is lost for that subscriber. This is at-most-once delivery, and nobody finds out.

Good: Call the handler first and move the offset only after it returns true. A failure leaves the offset in place, and the next drain offers the message again. This is at-least-once delivery. A handler can see the same message twice, so it must be safe to repeat (idempotent).

Great: Keep at-least-once and make repeats easy to handle. Each message has a unique offset, so a handler can remember the last offset it applied and skip repeats. Count the failed tries of a message. After a limit such as 3, move it to a dead-letter list and go on, so one bad message cannot stop a subscriber forever.

Trade-off: Exactly-once needs the handler's effect and the offset to be saved together in one transaction. That is costly and rarely worth it. At-least-once with an idempotent handler is the usual answer. State it in the contract so users know what to expect.

Unsubscribing while a delivery is running

Weak: Keep subscribers in an ArrayList and loop over it in drain. A handler that unsubscribes makes the loop throw ConcurrentModificationException. With an index loop the next subscriber is skipped instead, which is worse because nobody sees an error.

Good: Loop over a snapshot, for example a CopyOnWriteArrayList. Removing from the list no longer breaks the loop. But the snapshot still holds the removed subscriber, so it can receive one more message after unsubscribe returned.

Great: Use the snapshot and an active flag on each subscription. Unsubscribe sets the flag first, then removes the subscription. The delivery loop checks the flag before every message. The per-subscription lock is reentrant, so a handler can unsubscribe itself without a deadlock.

Trade-off: The flag is checked between messages. A handler that is already running cannot be stopped, so the contract says: no message starts after unsubscribe returns. Stopping a running handler needs interruption, which the handler must support.

Now add one more feature

Now add consumer groups: subscribers with the same group name share one offset, and each message goes to only one of them. What changes? Add a `ConsumerGroup` that owns the offset. A `Subscription` becomes a member that asks its group for the next message. The log and the Topic do not change.
Now limit memory: keep only the last 10,000 messages. What changes? The log becomes a ring or a list of segments, and offsets stay absolute. A subscriber that is behind the oldest kept offset gets an error or jumps forward, as the contract says.
Now let a subscriber receive only some messages. What changes? Give `Subscription` a `Predicate<Message>`. The offset still moves past messages that do not match, so the subscriber is not stuck on them.
Now survive a restart. What changes? Append each message to a file before returning the offset, and save each subscription's offset after a confirmation. On start, rebuild the log and the offsets. The Broker interface stays the same.

What each level is expected to show

SDE-1: Model Broker, Topic, Message and Subscriber, with Observer for fan-out. Show publish and subscribe, and say that a handler that throws must not stop the others. A plain list of handlers with a try and catch is acceptable.

SDE-2: Everything at SDE-1, plus a log with one offset per subscription and offsets that move only after confirmation. State a clear at-least-once contract. Isolate slow subscribers with a queue per subscriber or a pool with one task per subscription. Handle unsubscribe during a delivery with a snapshot and an active flag. A synchronous mode with drain keeps tests deterministic.

Common mistakes

Calling the handlers inside publish, so one slow subscriber slows the publisher and every other subscriber.
Moving the offset before the handler confirms. A failure then loses the message without any sign.
Keeping one shared offset for the topic instead of one per subscriber. Two subscribers cannot then read at different speeds.
Looping over the live subscriber list, so an unsubscribe from a handler throws or skips someone.
Promising exactly-once delivery without any way to make it true.
Letting a failed message block its subscriber forever. After a limit, park it in a dead-letter list.