Design an in-memory pub-sub system

“Design an in-memory pub-sub system: publishers send messages to topics, and subscribers to a topic receive them.” Then, almost always: “what happens when a subscriber is slow?”

Pub-sub is short for publish-subscribe. Code that has news (an order was placed) publishes a message to a named topic, like “orders”. Code that cares (the email sender, the warehouse) subscribes to that topic and receives every message. Neither side knows the other.

The first version is the Observer pattern, and it fits on a whiteboard: a map from topic to a list of handlers, and a publish that calls each one. The real question is what happens around it. Handlers are slow, they fail, they fall behind, and several threads publish at once. Each of those becomes a design decision, and the interviewer wants to hear each one named and made.

The framework

1. 01Answer
2. 02Scope
3. 03Build
4. 04Full design
5. 05Stress
6. 06Wrap up

The answer in one minute

If you only have a minute, this is the design. The rest of the page explains how we get here.

1. Each subscriber gets its own mailbox and its own worker. Publishing only drops the message into mailboxes and returns. So a slow or failing subscriber only delays itself.
2. Every mailbox has a size limit. When it’s full, the subscriber has chosen what happens: make the publisher wait, drop the oldest message, or drop the newest. A mailbox with no limit just means “out of memory, later”.
3. A failing message is retried, then set aside. After a few tries, it becomes a dead letter, kept for someone to inspect. One bad message can’t block a subscriber forever, and nothing disappears without a record.
4. One worker per subscriber keeps each subscriber’s order. Messages are handled one at a time, in the order they arrived.
5. Each topic numbers and hands out its messages one publish at a time. So when many threads publish, every subscriber still sees the same order.

We’ll build this in four stages, then deal with many threads publishing to one topic.

Scope it

Before drawing anything, we ask a few questions. Each answer changes a class. Anything that doesn’t, we assume and move on.

Interviewerminute 0

Design an in-memory pub-sub system.

What they are testing

Whether you ask about delivery guarantees and slow subscribers before drawing classes. “In memory, one process” rules out saving to disk and replaying, and candidates who ask spend their time on the parts that are actually being tested.

Here are the questions that matter:

• One process, or across machines? One process, in memory. Across machines is Kafka or a cloud queue, which is a system design question.
• Must messages survive a restart? No. Surviving a restart would need a log on disk.
• Do all subscribers need to see the same order? Yes, within a topic.
• What happens when a subscriber is slow? It’s the subscriber’s choice: slow the publisher down, or drop messages. One answer for everyone would be wrong for someone.
• Do late subscribers get old messages? No, only messages published after they subscribe.

These answers give us one promise that matters more than the others: a slow or failing subscriber never delays or breaks the others, or the publisher, beyond the limit it chose. Three smaller promises come with it. Every subscriber of a topic sees its messages in the same order. No subscriber can use unlimited memory. And closing the broker delivers what was already accepted, then stops.

We leave out saving to disk, replay, delivery across processes, and wildcard topics.

Go deeperThree more questions, and what each changes1 min

Question	We assume	If the answer were different
Wildcard topics (orders.*)?	No, exact names	Matching moves into the topic lookup; delivery doesn’t change
Message filters?	No	A check per subscription, before the mailbox
Delivery guarantee?	At most once on a crash, at least once on errors	Exactly once needs handlers that are safe to repeat, as everywhere else

Common mistake

Designing Kafka

Partitions, consumer groups, replication and offsets on disk are the answer to a different question. In an in-memory broker, they add classes without answering what the interviewer asks next: what happens to the publisher when one handler takes ten seconds?

Build it

We’ll build the design in four stages. Each stage adds only the classes that a new requirement needs. In each diagram, the new parts are outlined, and the colors show what each class does (the key is under every diagram).

Stage 1: A map from topic to handlers

Interviewerminute 3

Start with the simplest thing: an order service publishes “order placed”, and the email sender and the warehouse both need it. What are your classes?

What they are testing

Whether you reach the Observer shape in two minutes, and then name its problems yourself: the publisher’s thread runs every handler. The interviewer is about to ask about a slow handler, and it’s better if you get there first.

Start with the smallest thing that works. A Message has a topic, a payload, and a number that counts up within its topic, so a subscriber can tell if it missed one. A Subscriber is anything with an onMessage method. The Broker keeps a list of subscribers for each topic, and publish builds the message and calls each subscriber in turn. This is the Observer pattern: publishers don’t know who’s listening.

Stage 1: a broker that calls handlers

Scroll or pinch to zoom, drag to pan

• Color key
• Facade
• Entity
• Value object
• Enum
• Interface
• Implementation
• New in this step

JBroker.javaPybroker.pyC#Broker.csGobroker.goTSbroker.ts

JavaPythonC#GoTypeScript

JavaPythonC#GoTypeScript

// One published message. The sequence number is per topic, so a subscriber can tell whether it
// missed something or got something twice.
record Message(String topic, long seq, String payload) {}

// Whatever wants messages: an email sender, a cache invalidator, a metrics counter.
interface Subscriber {
    void onMessage(Message m);
}

class Broker {
    private final Map<String, List<Subscriber>> topics = new HashMap<>();
    private final Map<String, Long> nextSeq = new HashMap<>();

    void subscribe(String topic, Subscriber s) {
        topics.computeIfAbsent(topic, k -> new ArrayList<>()).add(s);
    }

    // flaw 1: the publisher's thread runs every handler. An email sender waiting on a mail
    // server slows every publish, and every subscriber after it waits too.
    // flaw 2: a handler that throws stops delivery to the subscribers after it, and the
    // exception lands in the publisher, which did nothing wrong.
    // flaw 3: nothing is safe across threads: two publishers can hand out the same sequence
    // number, and a subscribe during a publish breaks the loop. Stage 5.
    void publish(String topic, String payload) {
        long seq = nextSeq.merge(topic, 1L, Long::sum) - 1;
        Message m = new Message(topic, seq, payload);
        for (Subscriber s : topics.getOrDefault(topic, List.of())) s.onMessage(m);
    }
}

It works, but it has three problems, and we leave them in on purpose. The publisher’s own thread runs every handler. So an email sender waiting on a mail server slows every publish, and every subscriber after it waits too. A handler that throws an error stops delivery to the ones after it, and the error lands in the publisher’s lap. And nothing is safe when several threads publish at once.

Stage 2: A mailbox and a worker per subscriber

Interviewerminute 7

The email sender takes two seconds per message. The order service publishes fifty orders a second. What happens, and how do you fix it?

What they are testing

Whether you separate publishing from delivering. A thread pool shared by all subscribers is the common first answer. They’ll ask whether a subscriber’s messages can then arrive out of order, and what one slow subscriber does to the pool.

Let’s do the arithmetic. Fifty orders a second, two seconds each, means the email sender needs 100 seconds of work every second. With stage 1, the order service would grind to a halt.

So we separate publishing from delivering. Each subscription gets a mailbox (a queue) and one worker thread that takes messages out and calls the handler. Publishing only puts the message in each subscriber’s mailbox, and returns. The order service never waits on email, and the warehouse never waits behind the email sender. A queue between a fast producer and a slower consumer like this is called the producer-consumer pattern.

Why one worker per subscription, and not a shared pool of threads? A pool could handle two messages for the same subscriber at the same time. Then “order cancelled” could be handled before “order placed”. A single worker handles one message at a time, in the order they arrived.

A handler that throws now loses only that one message, which is counted, and the worker moves on. Closing a subscription puts a stop marker at the back of its mailbox. Everything accepted before the close is still delivered, and then the worker ends.

Stage 2: subscriptions with their own mailbox and worker

Scroll or pinch to zoom, drag to pan

• Color key
• Facade
• Entity
• Value object
• Enum
• Interface
• Implementation
• New in this step

JSubscription.javaPysubscription.pyC#Subscription.csGobroker.goTSsubscription.ts

JavaPythonC#GoTypeScript

JavaPythonC#GoTypeScript

// A subscriber's mailbox and the one thread that empties it. Publishing only drops the message
// in the mailbox, so the publisher never waits on a handler. One thread per subscription keeps
// each subscriber's messages in the order they were published, and one slow subscriber only
// delays itself.
class Subscription {
    private static final Message STOP = new Message("", -1, "");  // queued last by close()

    final String id;
    final String topic;
    private final Subscriber handler;
    private final BlockingQueue<Message> mailbox = new LinkedBlockingQueue<>();
    private final Thread worker;
    final AtomicLong failed = new AtomicLong();

    Subscription(String id, String topic, Subscriber handler) {
        this.id = id;
        this.topic = topic;
        this.handler = handler;
        this.worker = new Thread(this::run, "subscription-" + id);
        worker.start();
    }

    void offer(Message m) { mailbox.add(m); }

    // Close after delivering everything already in the mailbox: STOP goes in behind it, and the
    // worker exits when it gets there.
    void close() throws InterruptedException {
        mailbox.add(STOP);
        worker.join();
    }

    private void run() {
        try {
            for (Message m = mailbox.take(); m != STOP; m = mailbox.take()) deliver(m);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    // A handler that throws loses that message, not the subscription: the worker carries on
    // with the next one. Stage 4 retries it instead.
    private void deliver(Message m) {
        try {
            handler.onMessage(m);
        } catch (RuntimeException e) {
            failed.incrementAndGet();
        }
    }
}

class Broker {
    private final Map<String, List<Subscription>> topics = new HashMap<>();
    private final Map<String, Long> nextSeq = new HashMap<>();
    private final Supplier<String> ids;

    Broker(Supplier<String> ids) { this.ids = ids; }

    Subscription subscribe(String topic, Subscriber handler) {
        Subscription s = new Subscription(ids.get(), topic, handler);
        topics.computeIfAbsent(topic, k -> new ArrayList<>()).add(s);
        return s;
    }

    // Nothing new arrives after this; what's already in the mailbox is still delivered.
    void unsubscribe(Subscription s) throws InterruptedException {
        topics.getOrDefault(s.topic, new ArrayList<>()).remove(s);
        s.close();
    }

    // flaw: a mailbox has no limit. A subscriber that falls behind grows its mailbox until the
    // process runs out of memory, and nobody notices until then. Stage 3.
    void publish(String topic, String payload) {
        long seq = nextSeq.merge(topic, 1L, Long::sum) - 1;
        Message m = new Message(topic, seq, payload);
        for (Subscription s : topics.getOrDefault(topic, List.of())) s.offer(m);
    }

    void close() throws InterruptedException {
        for (List<Subscription> subs : topics.values()) for (Subscription s : subs) s.close();
    }
}

What does a thread per subscription cost? Ordinary threads don’t scale to tens of thousands of subscribers. But lightweight threads make it cheap: virtual threads in Java, goroutines in Go, and async loops in Node and C#.

Go deeperPublishing returns before anyone handles it1 min

The order service publishes. The broker puts the message in the email mailbox and the warehouse mailbox, and returns right away. Later, each worker takes its message out and calls its handler, at its own pace. The email worker might be two seconds behind, and nobody else notices.

Sequence: publishing returns before anyone handles it

Scroll or pinch to zoom, drag to pan

• Color key
• Background worker
• Facade

Go deeperFollowing one order through the system2 min

Let’s follow one message, start to finish, with two subscribers on “orders”: the email sender and the warehouse.

1. The order service calls publish("orders", order 41). The broker finds the “orders” topic and builds a message with number 41.
2. The broker puts message 41 at the back of the email mailbox, then at the back of the warehouse mailbox. Both are quick: adding to a queue takes microseconds.
3. publish returns. The order service goes on to order 42. Total time: a few microseconds, no matter how slow the handlers are.
4. The warehouse worker was idle, waiting on its empty mailbox. It wakes up, takes message 41 out, and calls the warehouse handler. That takes 20 milliseconds.
5. The email worker is still busy with message 39, which is waiting on the mail server. Messages 40 and 41 sit in its mailbox. When 39 finishes, the worker takes 40, then 41, in that order.

Nobody waits for anybody else. The only shared thing is the moment of adding to a mailbox, and that’s tiny. This is also why a mailbox needs a limit: in step 5, nothing stops the email mailbox from growing while the mail server is slow. That’s stage 3.

Stage 3: Size limits and backpressure

Interviewerminute 13

The email sender is down for an hour. Its mailbox now has 180,000 messages, and the process is using 4 GB. What should have happened?

What they are testing

Whether you know that a queue with no limit only delays the problem, and can name the choices for a full queue: block the producer, drop old, or drop new. Strong candidates say who should choose, and that drops must be counted.

Every mailbox now has a capacity, and each subscription chooses what happens when it’s full:

• Block: the publisher waits until there’s room. Nothing is lost, and the publisher slows down to the pace of the slowest blocking subscriber. This is called backpressure: the slowness shows up where the messages come from, instead of quietly turning into memory.
• Drop oldest: throw away the oldest waiting message to make room. Right for a price feed, where only the latest price matters.
• Drop newest: throw away the message being published. Right when the messages already waiting matter more, and new ones will come again.

Who should choose? The subscriber, because it knows what its messages mean. Every drop is counted on the subscription, so a dashboard can raise an alert when a subscriber starts losing messages.

Stage 3: offering a message to a full mailbox

Scroll or pinch to zoom, drag to pan

Stage 3: options per subscription

Scroll or pinch to zoom, drag to pan

• Color key
• Facade
• Entity
• Value object
• Enum
• Interface
• Implementation
• New in this step

JSubscription.javaPysubscription.pyC#Subscription.csGosubscription.goTSsubscription.ts

JavaPythonC#GoTypeScript

JavaPythonC#GoTypeScript

// What to do when a subscriber's mailbox is full. There is no right answer for every
// subscriber, so each one chooses when it subscribes.
enum Overflow {
    BLOCK,        // the publisher waits for room: nothing is lost, and a slow subscriber slows publishing
    DROP_OLDEST,  // make room by discarding the oldest waiting message: fine for "latest price" feeds
    DROP_NEWEST   // discard the message being published: the backlog is delivered in full
}

record Options(int capacity, Overflow overflow) {}

class Subscription {
    private static final Message STOP = new Message("", -1, "");

    final String id;
    final String topic;
    private final Subscriber handler;
    private final Overflow overflow;
    private final BlockingQueue<Message> mailbox;
    private final Thread worker;
    final AtomicLong failed = new AtomicLong();
    final AtomicLong dropped = new AtomicLong();  // a number someone can alert on

    Subscription(String id, String topic, Subscriber handler, Options options) {
        this.id = id;
        this.topic = topic;
        this.handler = handler;
        this.overflow = options.overflow();
        this.mailbox = new ArrayBlockingQueue<>(options.capacity());
        this.worker = new Thread(this::run, "subscription-" + id);
        worker.start();
    }

    void offer(Message m) throws InterruptedException {
        switch (overflow) {
            case BLOCK -> mailbox.put(m);
            case DROP_NEWEST -> {
                if (!mailbox.offer(m)) dropped.incrementAndGet();
            }
            case DROP_OLDEST -> {
                // The worker may take a message between the two calls, so loop until m fits.
                while (!mailbox.offer(m)) if (mailbox.poll() != null) dropped.incrementAndGet();
            }
        }
    }

    void close() throws InterruptedException {
        mailbox.put(STOP);
        worker.join();
    }

    private void run() {
        try {
            for (Message m = mailbox.take(); m != STOP; m = mailbox.take()) deliver(m);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    private void deliver(Message m) {
        try {
            handler.onMessage(m);
        } catch (RuntimeException e) {
            failed.incrementAndGet();
        }
    }
}

class Broker {
    private final Map<String, List<Subscription>> topics = new HashMap<>();
    private final Map<String, Long> nextSeq = new HashMap<>();
    private final Supplier<String> ids;

    Broker(Supplier<String> ids) { this.ids = ids; }

    Subscription subscribe(String topic, Subscriber handler, Options options) {
        Subscription s = new Subscription(ids.get(), topic, handler, options);
        topics.computeIfAbsent(topic, k -> new ArrayList<>()).add(s);
        return s;
    }

    void unsubscribe(Subscription s) throws InterruptedException {
        topics.getOrDefault(s.topic, new ArrayList<>()).remove(s);
        s.close();
    }

    // With BLOCK, this waits while any subscriber's mailbox is full. That is the backpressure:
    // the publisher slows to the pace of its slowest blocking subscriber.
    void publish(String topic, String payload) throws InterruptedException {
        long seq = nextSeq.merge(topic, 1L, Long::sum) - 1;
        Message m = new Message(topic, seq, payload);
        for (Subscription s : topics.getOrDefault(topic, List.of())) s.offer(m);
    }

    void close() throws InterruptedException {
        for (List<Subscription> subs : topics.values()) for (Subscription s : subs) s.close();
    }
}

Common mistake

Mailboxes with no limit, because nothing should be lost

A mailbox with no limit doesn’t prevent loss; it moves it. When the process runs out of memory, it loses every message in every mailbox, plus whatever else it was doing. A limited mailbox that blocks loses nothing either, and the slowness shows up where someone can see it.

Go deeperWith Block, can one subscriber stall the whole system?1 min

It can stall publishers to its own topic, and that’s the point: the system slows down to what it can deliver. It can’t stall other topics’ publishers. A subscriber that can’t afford to slow anyone down chooses a drop policy instead. A blocking publish can also take a time limit and fail, so the publisher decides how much a stalled subscriber may cost it.

Stage 4: Retries and dead letters

Interviewerminute 19

The warehouse handler calls a database that restarts now and then, so a few messages fail. One message has a broken payload and fails every time. Handle both.

What they are testing

Whether you tell brief failures from permanent ones, retry the first with growing waits, and set the second aside without blocking everything behind it. They also check whether retries keep the subscriber’s order.

Most failures are brief: a timeout, a restart, a busy lock. So a failing message is retried, up to a set number of attempts, waiting a little longer before each one. The worker retries the message right away, before taking the next one. That holds up this subscriber’s later messages during the retries, which is the price of keeping its order. A subscriber that doesn’t care about order could move failures to a side queue and carry on.

A message that fails every attempt becomes a dead letter. It’s set aside, together with the subscription and the error, and the worker moves on. Without this, one broken message would either be lost without a trace, or retried forever, blocking the subscriber for good. The broker keeps dead letters for a person or a repair job to look at and publish again.

Stage 4: delivering one message

Scroll or pinch to zoom, drag to pan

Stage 4: retry options and dead letters

Scroll or pinch to zoom, drag to pan

• Color key
• Facade
• Entity
• Value object
• Enum
• Interface
• Implementation
• New in this step

JSubscription.javaPysubscription.pyC#Subscription.csGosubscription.goTSsubscription.ts

JavaPythonC#GoTypeScript

JavaPythonC#GoTypeScript

// How one subscriber wants its messages delivered: how big its mailbox is, what to do when it
// is full, and how many times to retry a failing handler before giving up.
record Options(int capacity, Overflow overflow, int maxAttempts, long backoffMillis) {}

// A message a subscriber couldn't handle after every attempt, kept with the reason so a person
// or a repair job can look at it and publish it again.
record DeadLetter(Message message, String subscriptionId, String error) {}

// A subscriber's mailbox and the one worker that empties it. Publishing only drops the message
// in the mailbox, so the publisher never waits on a handler. One worker per subscription keeps
// each subscriber's messages in order, and a slow subscriber only delays itself.
class Subscription {
    // Queued last by close(), so the worker knows to stop.
    private static final Message STOP = new Message("", -1, "");

    final String id;
    final String topic;
    private final Subscriber handler;
    private final Options options;
    private final BlockingQueue<Message> mailbox;
    private final Consumer<DeadLetter> deadLetters;
    private final Thread worker;
    // Messages a full mailbox threw away. Counted, so a dashboard can show that this subscriber
    // is falling behind.
    final AtomicLong dropped = new AtomicLong();

    Subscription(String id, String topic, Subscriber handler, Options options, Consumer<DeadLetter> deadLetters) {
        this.id = id;
        this.topic = topic;
        this.handler = handler;
        this.options = options;
        this.mailbox = new ArrayBlockingQueue<>(options.capacity());
        this.deadLetters = deadLetters;
        this.worker = new Thread(this::run, "subscription-" + id);
        worker.start();
    }

    // Called by the topic while publishing. The subscriber's Overflow rule decides what a full
    // mailbox means: wait for room, drop the oldest message, or drop this one.
    void offer(Message m) throws InterruptedException {
        switch (options.overflow()) {
            case BLOCK -> mailbox.put(m);
            case DROP_NEWEST -> {
                if (!mailbox.offer(m)) dropped.incrementAndGet();
            }
            case DROP_OLDEST -> {
                while (!mailbox.offer(m)) if (mailbox.poll() != null) dropped.incrementAndGet();
            }
        }
    }

    // STOP goes in behind every waiting message, so the worker delivers those first and then
    // exits. join waits for that.
    void close() throws InterruptedException {
        mailbox.put(STOP);
        worker.join();
    }

    // The worker: one message at a time, in order, until STOP arrives.
    private void run() {
        try {
            for (Message m = mailbox.take(); m != STOP; m = mailbox.take()) deliver(m);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    // Most failures are brief (a database restarting, a timeout), so try again, waiting a
    // little longer each time. The next message waits meanwhile: that's the price of keeping
    // this subscriber's messages in order. After the last attempt the message is set aside as
    // a dead letter and the subscriber moves on, so one bad message can't stop it for good.
    private void deliver(Message m) throws InterruptedException {
        for (int attempt = 1; ; attempt++) {
            try {
                handler.onMessage(m);
                return;
            } catch (RuntimeException e) {
                if (attempt == options.maxAttempts()) {
                    deadLetters.accept(new DeadLetter(m, id, String.valueOf(e.getMessage())));
                    return;
                }
                Thread.sleep(options.backoffMillis() * attempt);
            }
        }
    }
}

class Broker {
    private final Map<String, List<Subscription>> topics = new HashMap<>();
    private final Map<String, Long> nextSeq = new HashMap<>();
    private final Queue<DeadLetter> deadLetters = new ConcurrentLinkedQueue<>();  // workers add from their own threads
    private final Supplier<String> ids;

    Broker(Supplier<String> ids) { this.ids = ids; }

    Subscription subscribe(String topic, Subscriber handler, Options options) {
        Subscription s = new Subscription(ids.get(), topic, handler, options, deadLetters::add);
        topics.computeIfAbsent(topic, k -> new ArrayList<>()).add(s);
        return s;
    }

    void unsubscribe(Subscription s) throws InterruptedException {
        topics.getOrDefault(s.topic, new ArrayList<>()).remove(s);
        s.close();
    }

    void publish(String topic, String payload) throws InterruptedException {
        long seq = nextSeq.merge(topic, 1L, Long::sum) - 1;
        Message m = new Message(topic, seq, payload);
        for (Subscription s : topics.getOrDefault(topic, List.of())) s.offer(m);
    }

    List<DeadLetter> deadLetters() { return List.copyOf(deadLetters); }

    void close() throws InterruptedException {
        for (List<Subscription> subs : topics.values()) for (Subscription s : subs) s.close();
    }
}

Go deeperHow long to wait between retries, and how many tries2 min

Our code waits backoff × attempt before each retry: with a backoff of 100 milliseconds, that’s 100, then 200, then 300. The waits grow, so a database that is restarting gets more time with each try, and a brief failure is retried quickly.

Why not retry at once, with no wait? Because the thing that failed is often overloaded. Retrying at full speed adds load at the worst moment, and every attempt fails the same way.

Many systems double the wait each time (100, 200, 400, 800), which is called exponential backoff, and add a small random amount to each wait, called jitter. Jitter matters when many workers fail at the same moment: without it, they all retry at the same moment too, and knock the service over again.

How many tries? Few. Three to five covers most brief failures. More tries mean a subscriber stuck for longer on a message that was never going to work, since every message behind it waits. The total time a message can spend in retries should be smaller than what the subscriber can afford to fall behind.

This is enough for a mid-level answer. Publishing is separate from delivering, with a mailbox per subscriber, the mailbox has a limit, and handler errors are caught. The rest of the page is what makes it a senior answer.

The full design

Here is everything together. Each class is here because a requirement needed it.

Pub-sub: full class diagram

Scroll or pinch to zoom, drag to pan

• Color key
• Facade
• Entity
• Value object
• Enum
• Interface
• Implementation

</>pub-sub· read only

JavaPythonC#GoTypeScript

JavaPythonC#GoTypeScript

Explorer

• JAllInOne.java
• src
• JBroker.java
• JDeadLetter.java
• JMessage.java
• JOptions.java
• JOverflow.java
• JSubscriber.java
• JSubscription.java
• JTopic.java

JAllInOne.java

AllInOne.javasrc/Broker.javasrc/DeadLetter.javasrc/Message.javasrc/Options.javasrc/Overflow.javasrc/Subscriber.javasrc/Subscription.javasrc/Topic.java

AllInOne.java

// The complete design, built from every stage on this page.

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
import java.util.function.*;

// One published message. The sequence number is per topic, so a subscriber can tell whether it
// missed something or got something twice.
record Message(String topic, long seq, String payload) {}

// Whatever wants messages: an email sender, a cache invalidator, a metrics counter.
interface Subscriber {
    void onMessage(Message m);
}

// The front door for publishers and subscribers. It keeps one Topic per name and a shared list
// of dead letters. It never runs a subscriber's code itself: each Subscription has its own
// worker, so one slow subscriber can't hold up the others.
class Broker {
    // Created on first use. Concurrent, because publishers and subscribers call in from
    // different threads.
    private final Map<String, Topic> topics = new ConcurrentHashMap<>();
    // Messages that failed every retry. Workers add to it from their own threads.
    private final Queue<DeadLetter> deadLetters = new ConcurrentLinkedQueue<>();
    // Injected, so tests get predictable subscription ids.
    private final Supplier<String> ids;

    Broker(Supplier<String> ids) { this.ids = ids; }

    // Each subscription gets its own mailbox and worker. Its failures go to the shared dead-
    // letter list.
    Subscription subscribe(String topic, Subscriber handler, Options options) {
        Subscription s = new Subscription(ids.get(), topic, handler, options, deadLetters::add);
        topic(topic).add(s);
        return s;
    }

    // Stop new messages first, then let the worker finish what is already in its mailbox.
    void unsubscribe(Subscription s) throws InterruptedException {
        topic(s.topic).remove(s);
        s.close();
    }

    // The topic numbers the message and puts it in every mailbox, so all subscribers see the
    // same order. A full mailbox can make this wait, depending on the subscriber's Overflow
    // rule.
    void publish(String topic, String payload) throws InterruptedException { topic(topic).publish(payload); }

    // A copy, so callers can't change the broker's list.
    List<DeadLetter> deadLetters() { return List.copyOf(deadLetters); }

    // Closes every subscription. Each close waits for its worker to finish.
    void close() throws InterruptedException {
        for (Topic t : topics.values()) for (Subscription s : t.subscriptions()) s.close();
    }

    private Topic topic(String name) { return topics.computeIfAbsent(name, Topic::new); }
}

// A subscriber's mailbox and the one worker that empties it. Publishing only drops the message
// in the mailbox, so the publisher never waits on a handler. One worker per subscription keeps
// each subscriber's messages in order, and a slow subscriber only delays itself.
class Subscription {
    // Queued last by close(), so the worker knows to stop.
    private static final Message STOP = new Message("", -1, "");

    final String id;
    final String topic;
    private final Subscriber handler;
    private final Options options;
    private final BlockingQueue<Message> mailbox;
    private final Consumer<DeadLetter> deadLetters;
    private final Thread worker;
    // Messages a full mailbox threw away. Counted, so a dashboard can show that this subscriber
    // is falling behind.
    final AtomicLong dropped = new AtomicLong();

    Subscription(String id, String topic, Subscriber handler, Options options, Consumer<DeadLetter> deadLetters) {
        this.id = id;
        this.topic = topic;
        this.handler = handler;
        this.options = options;
        this.mailbox = new ArrayBlockingQueue<>(options.capacity());
        this.deadLetters = deadLetters;
        this.worker = new Thread(this::run, "subscription-" + id);
        worker.start();
    }

    // Called by the topic while publishing. The subscriber's Overflow rule decides what a full
    // mailbox means: wait for room, drop the oldest message, or drop this one.
    void offer(Message m) throws InterruptedException {
        switch (options.overflow()) {
            case BLOCK -> mailbox.put(m);
            case DROP_NEWEST -> {
                if (!mailbox.offer(m)) dropped.incrementAndGet();
            }
            case DROP_OLDEST -> {
                while (!mailbox.offer(m)) if (mailbox.poll() != null) dropped.incrementAndGet();
            }
        }
    }

    // STOP goes in behind every waiting message, so the worker delivers those first and then
    // exits. join waits for that.
    void close() throws InterruptedException {
        mailbox.put(STOP);
        worker.join();
    }

    // The worker: one message at a time, in order, until STOP arrives.
    private void run() {
        try {
            for (Message m = mailbox.take(); m != STOP; m = mailbox.take()) deliver(m);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    // Most failures are brief (a database restarting, a timeout), so try again, waiting a
    // little longer each time. The next message waits meanwhile: that's the price of keeping
    // this subscriber's messages in order. After the last attempt the message is set aside as
    // a dead letter and the subscriber moves on, so one bad message can't stop it for good.
    private void deliver(Message m) throws InterruptedException {
        for (int attempt = 1; ; attempt++) {
            try {
                handler.onMessage(m);
                return;
            } catch (RuntimeException e) {
                if (attempt == options.maxAttempts()) {
                    deadLetters.accept(new DeadLetter(m, id, String.valueOf(e.getMessage())));
                    return;
                }
                Thread.sleep(options.backoffMillis() * attempt);
            }
        }
    }
}

// What to do when a subscriber's mailbox is full. There is no right answer for every
// subscriber, so each one chooses when it subscribes.
enum Overflow {
    BLOCK,        // the publisher waits for room: nothing is lost, and a slow subscriber slows publishing
    DROP_OLDEST,  // make room by discarding the oldest waiting message: fine for "latest price" feeds
    DROP_NEWEST   // discard the message being published: the backlog is delivered in full
}

// How one subscriber wants its messages delivered: how big its mailbox is, what to do when it
// is full, and how many times to retry a failing handler before giving up.
record Options(int capacity, Overflow overflow, int maxAttempts, long backoffMillis) {}

// A message a subscriber couldn't handle after every attempt, kept with the reason so a person
// or a repair job can look at it and publish it again.
record DeadLetter(Message message, String subscriptionId, String error) {}

// Two problems appear once several threads publish. Two publishers can read the same sequence
// number. And publisher A can put its message in Ana's mailbox, then publisher B puts its
// message in both mailboxes, then A reaches Ben's: Ana sees A, B and Ben sees B, A. Each topic
// therefore hands out the number and fills every mailbox as one step, under the topic's lock,
// so every subscriber sees the same order. Different topics never wait on each other.
//
// The subscriber list is copied on every change instead of locked. A publish that is blocked
// on a full mailbox holds the topic's lock, and an unsubscribe, often of that very subscriber,
// must not wait behind it.
class Topic {
    final String name;
    private final Object publishLock = new Object();
    private volatile List<Subscription> subscriptions = List.of();
    private long nextSeq;  // only touched under publishLock

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

    // The lock makes numbering the message and filling every mailbox one step.
    void publish(String payload) throws InterruptedException {
        synchronized (publishLock) {
            Message m = new Message(name, nextSeq++, payload);
            for (Subscription s : subscriptions) s.offer(m);  // a snapshot: changes don't affect this loop
        }
    }

    // Builds a new list with the change. A publish already looping over the old list is not
    // disturbed.
    synchronized void add(Subscription s) {
        List<Subscription> next = new ArrayList<>(subscriptions);
        next.add(s);
        subscriptions = List.copyOf(next);
    }

    // The same for removing. The lock here only stops two changes from racing; it never waits
    // on a publish.
    synchronized void remove(Subscription s) {
        List<Subscription> next = new ArrayList<>(subscriptions);
        next.remove(s);
        subscriptions = List.copyOf(next);
    }

    // Read without a lock: the list is never changed in place.
    List<Subscription> subscriptions() { return subscriptions; }
}

Read onlyJava 21Every file builds in CI9 files199 lines

Explorer

• Pyall_in_one.py
• pub_sub
• Pybroker.py
• Pydead_letter.py
• Pymessage.py
• Pyoptions.py
• Pyoverflow.py
• Pysubscription.py
• Pytopic.py

Pyall_in_one.py

all_in_one.pypub_sub/broker.pypub_sub/dead_letter.pypub_sub/message.pypub_sub/options.pypub_sub/overflow.pypub_sub/subscription.pypub_sub/topic.py

all_in_one.py

Read onlyPython 3.12Every file builds in CI8 files209 lines

Explorer

• C#AllInOne.cs
• src
• C#Broker.cs
• C#DeadLetter.cs
• C#Message.cs
• C#Options.cs
• C#Overflow.cs
• C#Subscription.cs
• C#Topic.cs

C#AllInOne.cs

AllInOne.cssrc/Broker.cssrc/DeadLetter.cssrc/Message.cssrc/Options.cssrc/Overflow.cssrc/Subscription.cssrc/Topic.cs

AllInOne.cs

Read onlyC# · .NET 10Every file builds in CI8 files207 lines

Explorer

• Goall_in_one.go
• Gobroker.go
• Godead_letter.go
• Gomessage.go
• Gooptions.go
• Gooverflow.go
• Gosubscriber.go
• Gosubscription.go
• Gotopic.go

Goall_in_one.go

all_in_one.gobroker.godead_letter.gomessage.gooptions.gooverflow.gosubscriber.gosubscription.gotopic.go

all_in_one.go

Read onlyGo 1.24Every file builds in CI9 files264 lines

Explorer

• TSall-in-one.ts
• src
• TSbroker.ts
• TSdead-letter.ts
• TSmessage.ts
• TSoptions.ts
• TSoverflow.ts
• TSsubscriber.ts
• TSsubscription.ts
• TStopic.ts

TSall-in-one.ts

all-in-one.tssrc/broker.tssrc/dead-letter.tssrc/message.tssrc/options.tssrc/overflow.tssrc/subscriber.tssrc/subscription.tssrc/topic.ts

all-in-one.ts

Read onlyTypeScript, strictEvery file builds in CI9 files208 lines

Three ideas hold it together:

• Publishing and delivering never share a thread. A publisher’s only contact with a subscriber is its mailbox (producer-consumer), and the mailbox’s policy decides how much a slow subscriber can cost.
• One reader per mailbox. That single worker is what gives each subscriber its order, and lets handlers be written without locks. Subscribers are plugged in through one interface (Observer).
• Nothing is lost silently. A message is delivered, dropped and counted, or set aside as a dead letter with its error. A set-aside store like this is called a dead letter queue.

Stress it

So far, one thread publishes at a time. But the order service runs eight threads, all publishing to “orders”.

Eight publishers, one topic

Interviewerminute 25

The order service runs eight threads, all publishing to “orders”. The email sender and the audit log both subscribe. Do they see the orders in the same order?

What they are testing

Whether you spot that each subscriber’s own order isn’t the same order for everyone, find the timing that breaks it, and fix it per topic without a global lock. They also listen for what an unsubscribe does while a publish is waiting.

Two things go wrong when several threads publish to one topic.

Two publishers can get the same number. Both read “the next number is 41” before either adds one.

The hand-out can interleave. Picture publishers A and B, each handing a message to two mailboxes:

1. A puts its message in the email mailbox.
2. B puts its message in the email mailbox, then in the audit mailbox.
3. A puts its message in the audit mailbox.

Email sees A then B. Audit sees B then A. Each subscriber’s order is still consistent with itself, but they disagree. And an audit log that disagrees with what was emailed is a bug report waiting to happen.

The fix is a lock per topic. Each topic numbers and hands out a message as one step, under the topic’s own lock. So publishes to one topic happen one at a time, and every subscriber sees messages in number order. Publishers to different topics use different locks, so they never wait for each other.

The subscriber list is the other shared thing. A publish that’s blocked on a full mailbox holds the topic’s lock. An unsubscribe, often of that very slow subscriber, must not wait behind it. So the list is never changed in place. Subscribe and unsubscribe build a new list and swap it in, and a publish goes through the list it started with. This is called copy on write.

Sequence: two publishers, without and with the topic lock

Scroll or pinch to zoom, drag to pan

Option	How	Same order for all?	Cost
1. A lock per topic for numbering and hand-out (our choice)	The subscriber list is a swapped copy	Yes	A blocked publish holds up other publishers to that topic
2. One global lock	Every publish takes it	Yes	Topics wait for each other for no reason
3. A queue per topic with its own dispatcher	Publishers add to the topic’s queue; one thread hands out	Yes	One more thread per topic, and one more queue to limit
4. No lock; subscribers sort by number	Subscribers reorder by sequence number themselves	Eventually	Every subscriber needs a reorder buffer and gap handling
5. Thread-safe collections only	A thread-safe map and lists	No	Safe from crashes, still interleaves

JBroker.javaPybroker.pyC#Topic.csGobroker.goTSbroker.ts

JavaPythonC#GoTypeScript

JavaPythonC#GoTypeScript

// Two problems appear once several threads publish. Two publishers can read the same sequence
// number. And publisher A can put its message in Ana's mailbox, then publisher B puts its
// message in both mailboxes, then A reaches Ben's: Ana sees A, B and Ben sees B, A. Each topic
// therefore hands out the number and fills every mailbox as one step, under the topic's lock,
// so every subscriber sees the same order. Different topics never wait on each other.
//
// The subscriber list is copied on every change instead of locked. A publish that is blocked
// on a full mailbox holds the topic's lock, and an unsubscribe, often of that very subscriber,
// must not wait behind it.
class Topic {
    final String name;
    private final Object publishLock = new Object();
    private volatile List<Subscription> subscriptions = List.of();
    private long nextSeq;  // only touched under publishLock

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

    // The lock makes numbering the message and filling every mailbox one step.
    void publish(String payload) throws InterruptedException {
        synchronized (publishLock) {
            Message m = new Message(name, nextSeq++, payload);
            for (Subscription s : subscriptions) s.offer(m);  // a snapshot: changes don't affect this loop
        }
    }

    // Builds a new list with the change. A publish already looping over the old list is not
    // disturbed.
    synchronized void add(Subscription s) {
        List<Subscription> next = new ArrayList<>(subscriptions);
        next.add(s);
        subscriptions = List.copyOf(next);
    }

    // The same for removing. The lock here only stops two changes from racing; it never waits
    // on a publish.
    synchronized void remove(Subscription s) {
        List<Subscription> next = new ArrayList<>(subscriptions);
        next.remove(s);
        subscriptions = List.copyOf(next);
    }

    // Read without a lock: the list is never changed in place.
    List<Subscription> subscriptions() { return subscriptions; }
}

// The front door for publishers and subscribers. It keeps one Topic per name and a shared list
// of dead letters. It never runs a subscriber's code itself: each Subscription has its own
// worker, so one slow subscriber can't hold up the others.
class Broker {
    // Created on first use. Concurrent, because publishers and subscribers call in from
    // different threads.
    private final Map<String, Topic> topics = new ConcurrentHashMap<>();
    // Messages that failed every retry. Workers add to it from their own threads.
    private final Queue<DeadLetter> deadLetters = new ConcurrentLinkedQueue<>();
    // Injected, so tests get predictable subscription ids.
    private final Supplier<String> ids;

    Broker(Supplier<String> ids) { this.ids = ids; }

    // Each subscription gets its own mailbox and worker. Its failures go to the shared dead-
    // letter list.
    Subscription subscribe(String topic, Subscriber handler, Options options) {
        Subscription s = new Subscription(ids.get(), topic, handler, options, deadLetters::add);
        topic(topic).add(s);
        return s;
    }

    // Stop new messages first, then let the worker finish what is already in its mailbox.
    void unsubscribe(Subscription s) throws InterruptedException {
        topic(s.topic).remove(s);
        s.close();
    }

    // The topic numbers the message and puts it in every mailbox, so all subscribers see the
    // same order. A full mailbox can make this wait, depending on the subscriber's Overflow
    // rule.
    void publish(String topic, String payload) throws InterruptedException { topic(topic).publish(payload); }

    // A copy, so callers can't change the broker's list.
    List<DeadLetter> deadLetters() { return List.copyOf(deadLetters); }

    // Closes every subscription. Each close waits for its worker to finish.
    void close() throws InterruptedException {
        for (Topic t : topics.values()) for (Subscription s : t.subscriptions()) s.close();
    }

    private Topic topic(String name) { return topics.computeIfAbsent(name, Topic::new); }
}

Why is this correct? Under the topic’s lock, a message gets its number and goes into every mailbox before the next publish can start. So the order of numbers is the order every mailbox receives them, and each worker hands them to its handler in that same order.

What does it cost? The topic lock is held while a publish waits for room in a blocking mailbox. That’s intended: publishers to that topic slow to its slowest blocking subscriber, which is what Block means.

Common mistake

A handler that publishes to its own topic

A handler that publishes back to the topic it listens on, with Block, can freeze forever: its worker waits for room in its own full mailbox, and only its worker can make room. Handlers that republish should use a different topic, or a drop policy. The broker can also spot a publish from a subscription’s own worker and refuse it.

Go deeperHow each language does the topic lock1 min

Each language uses a lock that can wait correctly. Java uses a monitor (synchronized with wait), and Go and Python use mutexes. C# uses a SemaphoreSlim, because a plain lock can’t be held across an await. TypeScript uses a chain of promises: in Node, every await is a point where another publish can run, so the chain makes each publish wait for the previous one to finish.

Go deeperWhy not the other options2 min

One global lock (option 2) is correct, and it’s a fine first answer. But one slow, blocking subscriber on “orders” would then stall publishers to “payments”, which have nothing to do with it. A lock per topic costs nothing more, so there’s no reason to share one.

A dispatcher thread per topic (option 3) is a real alternative, and it’s in the different designs at the end. Publishers no longer wait on the hand-out. But the topic queue needs its own limit and policy, so we’ve moved the backpressure question, not removed it.

Letting subscribers sort by number (option 4) is how systems spread across machines often work, because there’s no lock that every machine can share. Inside one process there is one, and it’s much simpler than asking every subscriber to buffer, reorder and handle gaps.

Thread-safe collections alone (option 5) are the trap. They stop the map and lists from being corrupted, so nothing crashes, and the tests that only count messages pass. But the order problem is about two steps, numbering and handing out, happening together. No collection can make two separate calls into one step. That needs a lock around both.

How you’d prove it works

Two tools make the tests reliable. close() waits until every worker has delivered what it accepted, so a test can publish, close, then check. And a “stuck” handler signals when it has started a message, then waits to be released. That lets a test fill a mailbox behind it at an exact moment. The tests that matter:

• Every subscriber gets all 100 messages, in order, and only its own topic’s messages.
• A slow subscriber doesn’t hold up another subscriber.
• A message that always fails becomes a dead letter, and the next message is still delivered. A message that fails once is retried and delivered.
• With capacity 2 and the worker stuck, drop-oldest delivers 1, 3, 4, and drop-newest delivers 1, 2, 3, each with one drop counted.
• With Block, a publisher waits while the mailbox is full, and finishes once there’s room.
• Unsubscribing stops new deliveries.
• Four threads publishing 250 messages each produce the same 1,000-message order for two subscribers.

Go deeperThe tests in five languages2 min

The last test catches the interleaving reliably in Java and Go. In Python and TypeScript, the window for the race is so small that the test may pass even without the topic lock. That’s a reminder that a passing concurrency test is evidence, not proof.

JStuck.javaPystuck.pyC#Stuck.csGostuck.goTSexample.ts

JavaPythonC#GoTypeScript

JavaPythonC#GoTypeScript

AtomicInteger nextId = new AtomicInteger();
Broker broker;

static final Options ROOMY = new Options(1000, Overflow.BLOCK, 3, 0);

@BeforeEach
void setUp() { broker = new Broker(() -> "s" + nextId.incrementAndGet()); }

// Collects payloads from the worker thread; read them only after close(), which joins it.
static class Inbox implements Subscriber {
    final List<String> got = Collections.synchronizedList(new ArrayList<>());
    public void onMessage(Message m) { got.add(m.payload()); }
}

// A handler that signals when it starts the first message and then waits to be released, so a
// test can fill the mailbox behind it.
static class Stuck extends Inbox {
    final CountDownLatch started = new CountDownLatch(1);
    final CountDownLatch release = new CountDownLatch(1);
    public void onMessage(Message m) {
        started.countDown();
        try {
            release.await();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        super.onMessage(m);
    }
}

void publishAll(String topic, String... payloads) throws InterruptedException {
    for (String p : payloads) broker.publish(topic, p);
}

@Test
void everySubscriberGetsEveryMessageInOrder() throws Exception {
    Inbox a = new Inbox(), b = new Inbox();
    broker.subscribe("orders", a, ROOMY);
    broker.subscribe("orders", b, ROOMY);
    for (int i = 0; i < 100; i++) broker.publish("orders", "o" + i);
    broker.close();
    assertEquals(100, a.got.size());
    assertEquals(a.got, b.got);
    assertEquals("o99", a.got.get(99));
}

@Test
void onlyTheTopicsSubscribersHearIt() throws Exception {
    Inbox orders = new Inbox(), prices = new Inbox();
    broker.subscribe("orders", orders, ROOMY);
    broker.subscribe("prices", prices, ROOMY);
    publishAll("orders", "o1");
    broker.close();
    assertEquals(List.of("o1"), orders.got);
    assertEquals(List.of(), prices.got);
}

@Test
void aSlowSubscriberDoesNotHoldUpTheOthers() throws Exception {
    Stuck slow = new Stuck();
    CountDownLatch fastGotIt = new CountDownLatch(1);
    broker.subscribe("orders", slow, ROOMY);
    broker.subscribe("orders", m -> fastGotIt.countDown(), ROOMY);
    publishAll("orders", "o1");
    assertTrue(fastGotIt.await(5, TimeUnit.SECONDS));  // while the slow one is still stuck
    slow.release.countDown();
    broker.close();
}

@Test
void aMessageThatKeepsFailingBecomesADeadLetter() throws Exception {
    Inbox ok = new Inbox();
    broker.subscribe("orders", m -> {
        if (m.payload().equals("bad")) throw new IllegalStateException("cannot parse");
        ok.onMessage(m);
    }, ROOMY);
    publishAll("orders", "o1", "bad", "o2");
    broker.close();
    assertEquals(List.of("o1", "o2"), ok.got);  // the bad one didn't stop the next
    DeadLetter d = broker.deadLetters().get(0);
    assertEquals("bad", d.message().payload());
    assertEquals("cannot parse", d.error());
}

@Test
void aBriefFailureIsRetried() throws Exception {
    AtomicInteger calls = new AtomicInteger();
    Inbox ok = new Inbox();
    broker.subscribe("orders", m -> {
        if (calls.incrementAndGet() == 1) throw new IllegalStateException("timeout");
        ok.onMessage(m);
    }, ROOMY);
    publishAll("orders", "o1");
    broker.close();
    assertEquals(List.of("o1"), ok.got);
    assertEquals(0, broker.deadLetters().size());
}

@Test
void dropOldestKeepsTheLatest() throws Exception {
    Stuck s = new Stuck();
    Subscription sub = broker.subscribe("prices", s, new Options(2, Overflow.DROP_OLDEST, 1, 0));
    publishAll("prices", "p1");
    s.started.await();  // the worker holds p1, the mailbox is empty
    publishAll("prices", "p2", "p3", "p4");
    s.release.countDown();
    broker.close();
    assertEquals(List.of("p1", "p3", "p4"), s.got);
    assertEquals(1, sub.dropped.get());
}

@Test
void dropNewestKeepsTheBacklog() throws Exception {
    Stuck s = new Stuck();
    Subscription sub = broker.subscribe("prices", s, new Options(2, Overflow.DROP_NEWEST, 1, 0));
    publishAll("prices", "p1");
    s.started.await();
    publishAll("prices", "p2", "p3", "p4");
    s.release.countDown();
    broker.close();
    assertEquals(List.of("p1", "p2", "p3"), s.got);
    assertEquals(1, sub.dropped.get());
}

@Test
void blockMakesThePublisherWait() throws Exception {
    Stuck s = new Stuck();
    broker.subscribe("orders", s, new Options(1, Overflow.BLOCK, 1, 0));
    publishAll("orders", "o1");
    s.started.await();
    publishAll("orders", "o2");  // fills the mailbox
    Thread publisher = new Thread(() -> {
        try {
            broker.publish("orders", "o3");
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    });
    publisher.start();
    publisher.join(200);
    assertTrue(publisher.isAlive());  // still waiting for room
    s.release.countDown();
    publisher.join();
    broker.close();
    assertEquals(List.of("o1", "o2", "o3"), s.got);
}

@Test
void unsubscribingStopsDelivery() throws Exception {
    Inbox a = new Inbox();
    Subscription sub = broker.subscribe("orders", a, ROOMY);
    publishAll("orders", "o1");
    broker.unsubscribe(sub);
    publishAll("orders", "o2");
    broker.close();
    assertEquals(List.of("o1"), a.got);
}

@Test
void concurrentPublishersProduceOneOrderForEveryone() throws Exception {
    Inbox a = new Inbox(), b = new Inbox();
    broker.subscribe("orders", a, ROOMY);
    broker.subscribe("orders", b, ROOMY);
    ExecutorService pool = Executors.newFixedThreadPool(4);
    List<Future<?>> done = new ArrayList<>();
    for (int t = 0; t < 4; t++) {
        int publisher = t;
        done.add(pool.submit(() -> {
            for (int i = 0; i < 250; i++) broker.publish("orders", publisher + "-" + i);
            return null;
        }));
    }
    for (Future<?> f : done) f.get();
    pool.shutdown();
    broker.close();
    assertEquals(1000, a.got.size());
    assertEquals(a.got, b.got);
}

Go deeperEvery edge case, and where it is handled2 min

Case	What happens	Where
Publish to a topic nobody subscribes to	Numbered, delivered to no one	Topic.publish
Subscriber added during a publish	Starts with the next message	Topic.add (copy)
Unsubscribe with messages still queued	They’re delivered, then the worker stops	Subscription.close
Handler throws	Retried, then a dead letter; the next message continues	Subscription.deliver
Mailbox full	The subscriber’s policy: wait, drop oldest, drop newest	Subscription.offer
Drops	Counted on the subscription	dropped
Close while messages are queued	Delivered first, then workers end	Broker.close
Handler republishes to its own topic	Can freeze with Block; use another topic	the handler’s choice
Process crashes	Queued messages are lost: this broker is in memory	out of scope

Wrap up

The decisions, and what each one costs

Decision	We chose	We gave up
Who runs handlers	A worker per subscription, never the publisher	A thread per subscriber, cheap only with light threads
Each subscriber’s order	One worker reading one mailbox	Handling one subscriber’s messages in parallel
Full mailboxes	A size limit, with block, drop-oldest or drop-newest per subscriber	One rule that’s simpler to explain
Failures	Retry in place with growing waits, then a dead letter	Later messages wait during retries
Same order for everyone	Number and hand out under a lock per topic	A blocked publish holds up its topic’s publishers
Subscribe and unsubscribe	A copied list, swapped on change	A copy of the list per change
Saving to disk	None: in memory, one process	Messages queued at a crash

Go deeperGenuinely different designs, and when to pick them2 min

A dispatcher thread per topic. Publishers add to a topic queue and return. One thread per topic hands messages out to subscriber mailboxes. Publishers never wait on the hand-out, and the order comes from the single dispatcher. It adds a queue to limit and a thread per topic. It suits many publishers on few topics.

Reactive streams. Libraries such as Reactor, RxJava and Akka Streams turn backpressure into a conversation: a subscriber asks for n messages, and the publisher sends no more until asked again. It’s the same idea as Block, written as a protocol, and it works across a whole chain of steps.

A synchronous event bus inside the application. Libraries like Guava’s EventBus deliver on the publisher’s thread by default, which is stage 1, on purpose. Inside one request, a synchronous handler that fails can undo the whole request. It suits events inside a database transaction, not slow side effects like email.

How to spend 45 minutes

• 0 to 3. One process, no saving to disk, same order per topic, and slow subscribers.
• 3 to 5. The promises. Name “a slow subscriber can’t hurt the others” as the hard one.
• 5 to 25. Handlers in a loop, a mailbox and worker per subscriber, size limits with policies, retries and dead letters. Code Subscription.offer and the worker loop.
• 25 to 27. Walk through the full diagram: publishing and delivering never share a thread.
• 27 to 34. Eight publishers on one topic: the interleaving, the topic lock, and the copied list.
• 34 to 45. Replay or consumer groups, then your own questions.

If time is short, skip retries and spend it on the overflow policies and the topic lock. Those are the two places this question is decided.

Follow-up questions

Interviewerminute 31

A new service starts up and wants the last hour of orders, not just new ones. What changes?

What they are testing

Whether you see that replay turns mailboxes into one shared log with a position per subscriber, which is how Kafka works, and what that costs in memory.

Most extensions fit into one place:

Extension	What changes	Touches delivery?
Replay from the last hour	A log per topic, a position per subscription	Yes
Wildcard topics (orders.*)	Topic lookup matches patterns; a subscription can belong to several topics	No
Message filters	A check per subscription, before the mailbox	No
Consumer groups (share the work)	A group shares one mailbox with several workers; order per key, not per group	Yes
Priorities	A priority queue as the mailbox	No
Metrics	Mailbox depth, drops and dead letters per subscription	No

Go deeperAdding replay, step by step2 min

Replay means messages must outlive delivery. Today, a message lives in a mailbox until its worker takes it out, then it’s gone. A new subscriber can’t see what’s already gone.

1. Each topic keeps a log. Every published message is added to the end of a list for its topic, and never changed. The log is trimmed from the front by age or size, for example “keep one hour”.
2. A subscription keeps a position, not a mailbox. The position (often called an offset) is the number of the next message it will read. Its worker reads from the log at its position, handles the message, and moves the position forward.
3. A new subscriber chooses where to start. “From now” sets its position to the end of the log. “From one hour ago” sets it to the first message still in the log. Nothing else is needed.
4. Memory changes shape. Slow subscribers no longer each hold their own backlog: they all read the same log, at different positions. Memory now depends on the log’s size, not on how far behind the slowest subscriber is.

There’s a new failure to report. If a subscriber falls so far behind that the trim removes messages it hasn’t read yet, those messages are lost to it. The broker should notice (the position is before the start of the log) and report it, just as it counts drops today.

This is exactly the idea behind Kafka. Kafka adds disks, many machines and copies of each log, but “a log per topic, and a position per reader” is the core.

Go deeperAdding consumer groups, step by step2 min

A consumer group is several workers sharing one subscription’s work. The warehouse runs four copies, and each order should go to just one of them, not all four.

1. The group shares one mailbox. Publishing to the topic still puts one copy into the group’s mailbox, as for any subscriber.
2. Several workers read from it. Four workers take messages out of the same mailbox. Each message is handled once, by whichever worker takes it.
3. Order is now per key, not per group. With four workers, message 41 and message 42 can be handled at the same time, so the group as a whole no longer sees one order. Usually only the order within one key matters, for example all messages for one customer.
4. To keep order per key, split the mailbox. Give the group four mailboxes, one per worker. Pick the mailbox by a hash of the key, so one customer’s messages always go to the same worker, in order. Kafka calls these partitions.

The overflow policy and dead letters stay as they are, per mailbox. The new cost is a hot key: if one customer sends half the orders, one worker gets half the work.

Go deeperFour more follow-up questions2 min

Interviewer asksWhat delivery guarantee does this broker give?

At least once for errors, since a failing handler is retried and may have partly done its work before failing. At most once for crashes, since queued messages live in memory. Handlers should be safe to repeat, using the topic and sequence number as a key. That’s what the number is for.

Interviewer asksHow would you shut down without losing messages?

Stop accepting publishes, then close each subscription, which delivers what’s queued and waits for the worker. A deadline puts a limit on it: anything still queued at the deadline is reported, so the operator knows what was lost. That’s what close() does, minus the deadline.

Interviewer asksWhy not use a thread pool for all subscribers?

Because two threads could handle the same subscriber’s messages at once, out of order, and the handler would need its own locking. A pool works if a subscriber runs on at most one thread at a time, which is the actor model. One worker per subscriber is the simplest way to get that, and with lightweight threads it’s cheap.

Interviewer asksHow do you find out a subscriber is struggling?

Three numbers per subscription: mailbox depth, drops, and dead letters. Depth climbing means it can’t keep up. Drops mean it’s losing messages, by its own choice. Dead letters mean its handler is failing. Each can raise its own alert.

Further reading

• Observer pattern, the shape of stage 1
• Design a logging framework, another producer-consumer queue with a full-queue policy
• Design a task scheduler, workers taking jobs from a queue
• Enterprise Integration Patterns: Dead Letter Channel, where dead letters come from
• Reactive Streams, backpressure as a protocol between publisher and subscriber