Design a Pub-Sub System

Design a Pub-Sub System

A quick but important distinction before starting: this problem is about designing an actual message broker — the infrastructure component that owns topics, subscriptions, and delivery guarantees, in the spirit of Kafka, SNS, or Redis Pub/Sub. It is not the same as the Observer/Pub-Sub design pattern covered elsewhere in this roadmap, where a subject directly holds references to its observers in-process with no persistence or delivery guarantees. Here, publishers and subscribers don't know about each other at all — they only know about topics — and the broker in the middle has to solve routing, persistence, and delivery semantics as first-class problems. That's what makes this a substantially bigger design than the pattern.

1. Requirements

Functional requirements

• Publishers publish messages to named topics.
• Subscribers subscribe to one or more topics and receive messages published after subscription (and, if persistent, potentially before).
• Support topic hierarchies with wildcard subscriptions: sports/* matches sports/cricket and sports/football (single-level), sports/** matches any depth including sports/cricket/live.
• Support delivery guarantees: at-least-once (default) and at-most-once (fire-and-forget, opt-in for latency-sensitive, loss-tolerant consumers).
• Support subscriber-side filtering by message attributes (e.g., only messages where region == "IN").
• Persist messages so that a subscriber that reconnects after being offline can still receive messages published during the gap, up to a retention policy.
• Route failed/unprocessable messages to a dead letter queue after a bounded number of delivery attempts.

Non-functional requirements

• Handle slow subscribers without blocking fast ones or the publisher (backpressure/buffering, not synchronous fan-out).
• Support high publish throughput with many topics and many subscribers per topic.
• Message ordering guaranteed per topic (or per partition, if partitioning is introduced), not globally across topics.
• Broker must survive a subscriber crash without losing messages that were in-flight but unacknowledged.

Out of scope: cross-datacenter replication, exactly-once semantics (a famously hard, often-debated guarantee even in production systems — worth naming as out of scope rather than hand-waving), a full persistent storage engine (a MessageStore interface is designed, but its backing implementation — disk-backed log, DB, etc. — is treated as pluggable).

2. Actors & Use Cases

Actors

• Publisher — publishes a Message to a Topic, unaware of who (if anyone) is subscribed.
• Subscriber — subscribes to a topic (possibly with a wildcard pattern and a filter), receives messages via a callback/handler, and acknowledges successful processing.
• Broker — owns topic registration, routing, persistence, and delivery-guarantee enforcement.
• Dead letter consumer — an operator or automated process that inspects messages that repeatedly failed delivery.

Primary use cases

1. A publisher publishes {"topic": "sports/cricket", "payload": {...}, "attributes": {"region": "IN"}} → broker matches it against all active subscriptions (exact and wildcard) → delivers to each matching subscriber's queue.
2. A subscriber subscribes to sports/* with a filter region == "IN" → only messages under any direct sports/X topic whose attribute matches are delivered to it; a message under sports/cricket/live (two levels deep) is not matched by a single-level wildcard.
3. A subscriber goes offline for 10 minutes, then reconnects → for a persistent subscription, the broker redelivers everything published since its last acknowledged offset, up to the retention window.
4. A subscriber processes a message but crashes before acknowledging → broker redelivers after a visibility timeout, since at-least-once semantics assume "no ack" means "not (yet) successfully processed."
5. A message fails delivery/processing 5 times → broker routes it to the dead letter queue and stops retrying, so one poison message can't block the topic indefinitely.
6. A slow subscriber's local queue fills up → broker applies backpressure (buffer up to a limit, then drop-oldest or disconnect the subscriber) rather than let one slow consumer stall the publisher or other subscribers.

3. Class Diagram

4. Core Class Design

Message and delivery types

public final class Message {
    private final String id;
    private final String topic;
    private final byte[] payload;
    private final Map<String, String> attributes;
    private final Instant timestamp;
 
    public Message(String topic, byte[] payload, Map<String, String> attributes) {
        this.id = UUID.randomUUID().toString();
        this.topic = topic;
        this.payload = payload;
        this.attributes = Map.copyOf(attributes);
        this.timestamp = Instant.now();
    }
    public String id() { return id; }
    public String topic() { return topic; }
    public byte[] payload() { return payload; }
    public Map<String, String> attributes() { return attributes; }
}
 
public enum DeliveryMode { AT_LEAST_ONCE, AT_MOST_ONCE }
public enum Ack { SUCCESS, RETRY, DEAD_LETTER }
 
public interface Subscriber {
    String id();
    Ack onMessage(Message message);
}
 
public interface MessageFilter {
    boolean matches(Message message);
}
 
public final class AttributeFilter implements MessageFilter {
    private final String key;
    private final String expectedValue;
    public AttributeFilter(String key, String expectedValue) { this.key = key; this.expectedValue = expectedValue; }
    public boolean matches(Message message) {
        return expectedValue.equals(message.attributes().get(key));
    }
}

Topic trie — routes exact and wildcard subscriptions in O(depth), not O(subscriptions)

public final class TopicTrie {
    private final TrieNode root = new TrieNode();
 
    public void register(String pattern, Subscription subscription) {
        String[] segments = pattern.split("/");
        TrieNode node = root;
        for (String segment : segments) {
            node = node.children.computeIfAbsent(segment, s -> new TrieNode());
        }
        node.subscriptions.add(subscription);
    }
 
    public List<Subscription> match(String topicName) {
        String[] segments = topicName.split("/");
        List<Subscription> matches = new ArrayList<>();
        collect(root, segments, 0, matches);
        return matches;
    }
 
    private void collect(TrieNode node, String[] segments, int i, List<Subscription> matches) {
        if (i == segments.length) {
            matches.addAll(node.subscriptions);
            // A "**" child at this exact point also matches "zero remaining segments"
            TrieNode multi = node.children.get("**");
            if (multi != null) matches.addAll(multi.subscriptions);
            return;
        }
        // Exact segment match
        TrieNode exact = node.children.get(segments[i]);
        if (exact != null) collect(exact, segments, i + 1, matches);
 
        // Single-level wildcard "*" matches exactly this one segment
        TrieNode single = node.children.get("*");
        if (single != null) collect(single, segments, i + 1, matches);
 
        // Multi-level wildcard "**" matches this segment and any number of following ones
        TrieNode multi = node.children.get("**");
        if (multi != null) {
            matches.addAll(multi.subscriptions);          // matches remaining path of any length
            collect(multi, segments, i + 1, matches);      // also allow "**" to be followed by more literal structure
        }
    }
 
    private static final class TrieNode {
        Map<String, TrieNode> children = new HashMap<>();
        List<Subscription> subscriptions = new ArrayList<>();
    }
}

Broker

public final class PubSubBroker {
    private final Map<String, Topic> topics = new ConcurrentHashMap<>();
    private final TopicTrie subscriptionIndex = new TopicTrie();
    private final MessageStore store;
    private final Map<String, DeliveryWorker> workersBySubscription = new ConcurrentHashMap<>();
 
    public PubSubBroker(MessageStore store) { this.store = store; }
 
    public void createTopic(String name) {
        topics.computeIfAbsent(name, Topic::new);
    }
 
    public Subscription subscribe(String topicPattern, Subscriber subscriber, MessageFilter filter, DeliveryMode mode) {
        Subscription subscription = new Subscription(topicPattern, subscriber, filter, mode);
        subscriptionIndex.register(topicPattern, subscription);
        DeliveryWorker worker = new DeliveryWorker(subscription, store);
        workersBySubscription.put(subscription.id(), worker);
        worker.start();
        return subscription;
    }
 
    public void publish(String topicName, Message message) {
        topics.computeIfAbsent(topicName, Topic::new);
        long offset = store.append(topicName, message); // durability before fan-out
 
        List<Subscription> matched = subscriptionIndex.match(topicName);
        for (Subscription subscription : matched) {
            if (!subscription.filter().matches(message)) continue;
            workersBySubscription.get(subscription.id()).enqueue(message, offset);
        }
    }
 
    public void unsubscribe(Subscription subscription) {
        DeliveryWorker worker = workersBySubscription.remove(subscription.id());
        if (worker != null) worker.stop();
    }
}

5. Design Patterns Applied

6. Key Algorithms, Concurrency & Edge Cases

Wildcard topic matching

The TopicTrie above is the centerpiece: naively checking every subscription's pattern against every published message is O(subscriptions) per publish; the trie makes matching O(topic depth × branching factor), which for realistic topic hierarchies (a handful of segments) is effectively constant time regardless of how many subscriptions exist.

At-least-once delivery: ack, retry, and visibility timeout

public final class DeliveryWorker {
    private final Subscription subscription;
    private final MessageStore store;
    private final BlockingQueue<Delivery> queue = new LinkedBlockingQueue<>(1000); // bounded: backpressure point
    private final int maxRetries = 5;
    private final DeadLetterQueue dlq;
    private volatile boolean running = true;
    private Thread thread;
 
    public void enqueue(Message message, long offset) {
        boolean offered = queue.offer(new Delivery(message, offset, 0));
        if (!offered) {
            // Queue full: this subscriber is falling behind. For AT_LEAST_ONCE we rely on the
            // durable MessageStore as the source of truth and simply drop this in-memory hint —
            // the subscriber will catch up by re-reading from its last acked offset.
        }
    }
 
    void run() {
        while (running) {
            try {
                Delivery delivery = queue.poll(500, TimeUnit.MILLISECONDS);
                if (delivery == null) continue;
 
                Ack result = invokeWithTimeout(delivery.message());
 
                switch (result) {
                    case SUCCESS -> store.ack(subscription, delivery.offset());
                    case RETRY -> {
                        if (delivery.attempts() + 1 >= maxRetries) {
                            dlq.add(delivery.message(), subscription, "max retries exceeded");
                            store.ack(subscription, delivery.offset()); // advance past the poison message
                        } else {
                            // Exponential backoff before redelivery
                            long backoffMs = (long) Math.pow(2, delivery.attempts()) * 100;
                            scheduleRedelivery(delivery.retry(), backoffMs);
                        }
                    }
                    case DEAD_LETTER -> {
                        dlq.add(delivery.message(), subscription, "subscriber-requested dead letter");
                        store.ack(subscription, delivery.offset());
                    }
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
 
    private Ack invokeWithTimeout(Message message) {
        try {
            return subscription.subscriber().onMessage(message);
        } catch (Exception e) {
            return Ack.RETRY; // an uncaught exception is treated as "not successfully processed"
        }
    }
 
    // start(), stop(), scheduleRedelivery() elided for brevity
    private record Delivery(Message message, long offset, int attempts) {
        Delivery retry() { return new Delivery(message, offset, attempts + 1); }
    }
}

At-least-once vs. at-most-once, precisely: at-least-once only advances the acknowledged offset after a successful onMessage (or after exhausting retries into the DLQ) — a crash mid-processing means redelivery, so subscriber handlers must be idempotent. At-most-once advances the offset (or simply never persists/retries) before or without confirming processing — faster and simpler, but a crash mid-processing silently loses that message. The choice is a deliberate trade a subscriber opts into, not a broker-wide constant.

Persistence and catch-up delivery

MessageStore.append returns a monotonically increasing per-topic offset; a subscription tracks lastAckedOffset. A reconnecting subscriber's DeliveryWorker calls store.readFrom(topic, lastAckedOffset + 1) to replay everything missed — this is the same offset-based replay model Kafka consumers use, and naming that explicitly signals familiarity with production systems.

Backpressure for slow subscribers

Each subscription's DeliveryWorker has its own bounded queue (capacity=1000 above) — a slow subscriber fills only its own queue, never blocking the publisher (which writes to the durable MessageStore and returns) or other subscribers' independent queues. This isolation is the entire reason the design fans out to per-subscription workers instead of one shared dispatch loop.

Edge cases

• Duplicate delivery under at-least-once: expected and must be handled by the subscriber via idempotency (e.g., dedupe by message.id()), not eliminated by the broker — eliminating it entirely is the "exactly-once" problem, explicitly out of scope.
• Ordering across a topic with multiple subscribers reading concurrently: order is preserved within the MessageStore's append log per topic; whether a slow subscriber processes messages 1, 2, 3 strictly in order depends on the DeliveryWorker being single-threaded per subscription (as designed above) — a thread pool per subscription would need explicit sequencing to preserve order.
• A subscriber unsubscribes while messages are in-flight in its queue: unsubscribe should stop the worker cleanly, and in-flight-but-unacked messages simply remain unacked in the store, ready for a future resubscription to pick up from the last acked offset.
• Wildcard subscription registered after messages already published: only affects future publishes; if catch-up from a retention window is desired, the new subscription's lastAckedOffset should be initialized to the earliest retained offset, not "now."

7. Trade-offs & Extensions

Natural extensions:

• Partitioning: split a high-throughput topic into partitions, each independently ordered, to parallelize delivery — the step from "in-process broker" toward "Kafka."
• Consumer groups: multiple subscriber instances sharing one logical subscription, each partition delivered to exactly one group member — enables horizontal scaling of consumption.
• Schema validation: reject or route to DLQ messages that don't conform to a registered schema for the topic.
• Priority topics: some topics/subscriptions processed ahead of others under load — requires a priority queue instead of a plain FIFO per worker.

Interview Questions

• How is this "Pub-Sub system" design different from just implementing the Observer pattern? What does the broker add that a subject-with-observer-list doesn't have?
• Walk through how sports/* and sports/** differ in what they match, and how the trie-based matcher implements that difference.
• Explain at-least-once vs. at-most-once delivery precisely — what does each guarantee, and what must subscriber code do differently under each?
• How does a reconnecting subscriber catch up on messages it missed while offline, and what state does the broker need to track to make that possible?
• Why does each subscription get its own delivery queue instead of one shared dispatch queue for the whole broker? What failure mode does that isolation prevent?
• How would you extend this design to guarantee ordering only within a topic but allow parallel delivery across topics?
• What happens to a message that fails processing 5 times in a row, and why is that better than retrying forever?