Design a Pub-Sub Message Broker

Design a Notification/Observer-Based Pub-Sub built an in-process event bus. This lesson is a step up in scope — a standalone broker component, still in-memory, but modeling the actual broker/topic/producer/consumer shape a real message queue uses.

Core classes

class Message { String id; String payload; Instant timestamp; }

class Topic {
    private final Queue<Message> messages = new ConcurrentLinkedQueue<>();
    private final List<Subscriber> subscribers = new CopyOnWriteArrayList<>();
}

interface Subscriber { void onMessage(Message message); }

class Publisher {
    private final Broker broker;
    void publish(String topicName, String payload) {
        broker.getTopic(topicName).deliver(new Message(UUID.randomUUID().toString(), payload, Instant.now()));
    }
}

class Broker {
    private final Map<String, Topic> topics = new ConcurrentHashMap<>();
    Topic getTopic(String name) { return topics.computeIfAbsent(name, k -> new Topic()); }
}

Basic in-memory pub-sub with topic-based routing

class Topic {
    // ...fields above...
    void subscribe(Subscriber subscriber) { subscribers.add(subscriber); }

    void deliver(Message message) {
        for (Subscriber s : subscribers) {
            s.onMessage(message); // synchronous fan-out to every current subscriber of this topic
        }
    }
}

Topic-based routing here is deliberately simple — a message published to "orders.created" only reaches subscribers of exactly that topic name, with no pattern matching or hierarchical routing (a real broker's "orders.*" wildcard subscriptions, for instance) — a reasonable, explicitly-scoped simplification for an LLD exercise.

What's missing vs a real broker

Naming these gaps explicitly is the actual point of this exercise — recognizing the difference between "a working pub-sub toy" and "a production message queue" is a stronger signal than presenting this simplified version as complete:

• Persistence: this Topic's messages exist only in a ConcurrentLinkedQueue in one process's memory — a process restart loses everything undelivered. Kafka/RabbitMQ durably write messages to disk before acknowledging a publish.
• Partitioning: this design has one queue per topic, on one machine — Kafka partitions a topic across multiple machines specifically for horizontal write/read throughput (see Design: Message Queue System for the full partition/replication model).
• Delivery guarantees: deliver() here is fire-and-forget synchronous iteration — no acknowledgment, no retry on a subscriber failure, no offset tracking to resume from a specific point after a consumer restart. Real brokers offer at-least-once (or exactly-once, with more machinery) delivery guarantees backed by consumer acknowledgment and offset commits.

From push-to-callbacks to a log with consumer offsets

The in-memory broker above pushes each message to subscriber callbacks as it arrives. That has three structural limits: a slow subscriber slows the publisher, a subscriber that's down misses messages forever, and nothing can be replayed. Modern brokers such as Kafka solve all three with one idea: store messages in an append-only log per topic, and let each consumer group track its own position (offset) in that log.

final class TopicLog {
    private final List<Message> entries = new ArrayList<>();       // append-only; index = offset
    private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();

    long append(Message m) {
        lock.writeLock().lock();
        try { entries.add(m); return entries.size() - 1; }
        finally { lock.writeLock().unlock(); }
    }

    List<Message> read(long fromOffset, int max) {
        lock.readLock().lock();
        try {
            int from = (int) Math.min(fromOffset, entries.size());
            return List.copyOf(entries.subList(from, Math.min(entries.size(), from + max)));
        } finally { lock.readLock().unlock(); }
    }
}

final class Broker {
    private final Map<String, TopicLog> topics = new ConcurrentHashMap<>();
    private final Map<String, AtomicLong> committed = new ConcurrentHashMap<>();   // "group|topic" -> next offset

    long publish(String topic, Message m) {
        return topics.computeIfAbsent(topic, t -> new TopicLog()).append(m);
    }

    List<Message> poll(String group, String topic, int max) {
        long from = committed.computeIfAbsent(group + "|" + topic, k -> new AtomicLong()).get();
        return topics.getOrDefault(topic, new TopicLog()).read(from, max);
    }

    void commit(String group, String topic, long nextOffset) {
        committed.get(group + "|" + topic).accumulateAndGet(nextOffset, Math::max);   // never move backwards
    }
}

The consumer side becomes a pull loop:

while (running) {
    List<Message> batch = broker.poll("billing", "orders", 100);
    for (Message m : batch) handler.handle(m);                         // process first…
    if (!batch.isEmpty()) broker.commit("billing", "orders", lastOffset(batch) + 1);   // …then commit
    if (batch.isEmpty()) Thread.sleep(50);                             // or long-poll
}

What this model gives you:

• Decoupled speeds: publishers only append. Each consumer group reads at its own pace, and a slow group builds up lag (committed offset versus log end), which is easy to monitor, instead of blocking anyone.
• Independent consumer groups: "billing" and "analytics" each receive every message, with their own offsets. That's pub-sub fan-out without the broker copying messages.
• Replay: reset a group's offset to reprocess history after a bug fix, or to bootstrap a new service.
• Delivery semantics are explicit in the code. Processing then committing gives at-least-once delivery: a crash between the two means the batch is re-read after restart. Committing before processing gives at-most-once. Handlers must therefore be idempotent (deduplicate by message ID) for at-least-once to be safe.

Scaling a topic: partitions

A single log is consumed in order, by one reader per group. To parallelize, split a topic into partitions, each its own log. Publishers choose a partition by hash(key) % partitions, so all messages for one key (one order ID, one user) stay in the same partition, in order. Within a group, each partition is assigned to exactly one consumer instance, so ordering per key is preserved while different keys are processed in parallel. The maximum parallelism of a group equals the number of partitions.

Retention and dead letters

• Retention: logs can't grow forever. Delete segments older than N days, or keep only the latest value per key (log compaction). Consumers whose committed offset has fallen behind the retention window must be reset, and this should be monitored.
• Poison messages: a message that always fails would block its partition forever under at-least-once delivery. After a few retries, move it to a dead-letter topic with the error attached, commit past it, and alert, so one bad message doesn't stop all processing.

Follow-up questions this topic invites — and their answers

Q: If you had to add ONE of these three missing features first for a real use case, which, and why? A: Usually persistence — an in-memory-only broker that loses all undelivered messages on restart is disqualifying for most real production use cases immediately, while partitioning and refined delivery guarantees are scaling/robustness refinements that matter once the basic durability requirement is already met.

Q: How would you add at-least-once delivery to this design without a full broker rewrite? A: Track per-subscriber acknowledgment (a subscriber calls back to confirm processing) and retain a message in the Topic's queue until every subscriber has acknowledged it, retrying delivery to any subscriber that hasn't acknowledged within a timeout — a meaningful step up in complexity from the current fire-and-forget deliver(), but a natural extension rather than a redesign.

Q: Is synchronous deliver() (subscribers processed in a loop on the publishing thread) a design smell? A: For an LLD exercise it's an acceptable simplification, but worth naming: a slow subscriber blocks the publisher (and every other subscriber's delivery) under this design — the same sync-vs-async tradeoff from Design a Notification/Observer-Based Pub-Sub applies, and a production version would dispatch to subscribers asynchronously (an executor per subscriber, or per topic) to decouple their processing time from the publisher's.

Q: How does ConcurrentLinkedQueue's choice here compare to using a plain synchronized Queue? A: ConcurrentLinkedQueue is a lock-free (CAS-based) concurrent queue implementation, generally offering better throughput under contention than externally synchronizing a plain LinkedList — a reasonable default choice for a queue accessed by potentially many publisher and consumer threads concurrently, without needing to hand-roll the synchronization Producer-Consumer Class Design builds explicitly for its own pedagogical purpose.

Q: Why do log-based brokers let consumers pull instead of pushing to them? A: Pulling lets each consumer control its own rate and batch size, so slow consumers build up lag instead of being overwhelmed or slowing the broker. It also makes replay trivial: a consumer just reads from an earlier offset.

Q: How are ordering and parallelism balanced with partitions? A: Messages with the same key go to the same partition, and each partition is consumed by one member of a group at a time, so per-key order is preserved. Different partitions are processed in parallel, so throughput scales with the partition count, but there's no total order across partitions.

Q: Why does at-least-once delivery require idempotent consumers? A: If a consumer processes a message but crashes before committing its offset, the message is delivered again after restart. Idempotent handling (deduplicating by message ID, or using naturally idempotent writes) makes that duplicate harmless.