Why this matters: three pieces of code carry this round — the publish path that appends under the topic's sequencing lock and returns, the enqueue that implements the bounded-drop contract, and the worker loop that drains in order and handles both delivery modes. Interviewers don't need compiling thread primitives; they need disciplined pseudocode that names what is locked and when. That's exactly what this walkthrough writes.
Vocabulary types first
record Message(long seq, String topic, Object payload) {}
// seq is assigned by the topic at publish — the order IS the design's
// per-topic promise, so it's stamped at the one place order exists.
enum DeliveryMode { AT_MOST_ONCE, AT_LEAST_ONCE }
record SubscriptionConfig(DeliveryMode mode, int queueCapacity) {}Narrate the seq comment: the sequence number is born at the topic's sequencing point, so the ordering promise has a physical location in the code.
The subscription: queue, worker, mode — one owner
class Subscription {
final String id;
final Consumer<Message> callback;
final SubscriptionConfig config;
final Deque<Message> queue = new ArrayDeque<>(); // guarded by 'this'
final Map<Long, Message> pending = new HashMap<>(); // at-least-once only
long dropped = 0; // the drop counter
volatile boolean active = true;
// one worker thread drains this queue — created on subscribe
synchronized void enqueue(Message m) {
if (queue.size() == config.queueCapacity()) {
queue.pollFirst(); // full: drop the OLDEST, per contract
dropped++; // counted, never silent
}
queue.addLast(m);
notify(); // wake the worker
}
}Say the contract while writing enqueue: "capacity is per-subscription config; when full, the oldest goes, and we count it — a lagging subscriber loses its oldest backlog first, and that's a fact we can report, not a mystery." Note what's locked: the queue is guarded by the subscription's own monitor — per-subscription state, per-subscription lock, so contention never crosses subscribers.
The topic: sequencer plus copy-on-write list
class Topic {
final String name;
private final AtomicLong nextSeq = new AtomicLong();
// copy-on-write: publish snapshots; subscribe/unsubscribe swap
private volatile List<Subscription> subs = List.of();
private final Object publishLock = new Object();
void publish(Object payload) {
synchronized (publishLock) { // THE per-topic ordering point
Message m = new Message(nextSeq.getAndIncrement(), name, payload);
for (Subscription s : subs) // snapshot via volatile read
s.enqueue(m);
} // lock covers seq assignment + fan-out enqueue: no interleaving,
// so every queue sees topic order. NO callback runs in here.
}
synchronized void add(Subscription s) { // copy-on-write swap
var next = new ArrayList<>(subs); next.add(s);
subs = List.copyOf(next);
}
synchronized void remove(Subscription s) {
var next = new ArrayList<>(subs); next.remove(s);
subs = List.copyOf(next);
}
}Two narration beats, both about what the lock covers. First: the publish lock spans sequence assignment and the fan-out loop — two concurrent publishes can't interleave their enqueues, so every subscriber's queue receives messages in topic order. Second, the boundary-race answer: a subscribe that lands mid-publish swapped the list after this publish snapshotted it, so the new subscriber cleanly misses this message and sees the next — never half the fan-out, never twice. The lock's critical section is a fan-out of cheap enqueues — no callback, no I/O — which is what makes holding it acceptable.
The delivery worker: drain in order, mode decides the tail
// one per subscription, started on subscribe
void workerLoop(Subscription s) {
while (s.active) {
Message m = s.takeNext(); // waits on the monitor when empty
try {
s.callback.accept(m);
if (s.config.mode() == DeliveryMode.AT_LEAST_ONCE)
s.pending.put(m.seq(), m); // held until ack(seq)
} catch (Exception e) {
if (s.config.mode() == DeliveryMode.AT_LEAST_ONCE)
s.enqueue(m); // redeliver — duplicates possible
// AT_MOST_ONCE: loss, exactly as contracted
}
}
}
void ack(Subscription s, long seq) {
s.pending.remove(seq); // the only exit from pending
}This loop is where the guarantees become visible, so narrate each branch: the callback runs here, on the subscription's own worker — never on a publisher's thread; one worker draining one queue is why per-subscriber order holds; and the catch block is the delivery contract in four lines — at-least-once re-enqueues (say it: "the subscriber may now see this message twice — that's the contract they chose"), at-most-once accepts the loss.
The broker: thin by design
class Broker {
private final ConcurrentHashMap<String, Topic> topics = new ConcurrentHashMap<>();
void publish(String topic, Object payload) {
topics.computeIfAbsent(topic, Topic::new).publish(payload);
}
Subscription subscribe(String topic, Consumer<Message> cb,
SubscriptionConfig cfg) {
Subscription s = new Subscription(newId(), cb, cfg);
startWorker(s);
topics.computeIfAbsent(topic, Topic::new).add(s);
return s;
}
void unsubscribe(String topic, Subscription s) {
topics.get(topic).remove(s); // 1: no new enqueues after this
s.active = false; // 2: worker exits after current message
s.wakeWorker(); // (never mid-callback)
}
}The broker holds almost nothing — computeIfAbsent gives topics-on-first-use, and all real behavior lives where its state lives. Unsubscribe's two ordered steps deserve a sentence: remove from the list first so nothing new arrives, then stop the worker at a message boundary — a callback is never killed halfway.
What you'd say about complexity
Publish: O(s) enqueues for s subscribers on the topic, under the topic lock — each enqueue is O(1) deque work, so the critical section stays microseconds even at hundreds of subscribers. Delivery: O(1) per message per subscriber, on the subscriber's own worker. Memory: bounded by design — the sum of queue capacities is the broker's worst case, which is the point of bounded queues. The honest caveat: the per-topic lock serializes publishers on that topic — fine because the contract only promises order per topic, and hot topics pay for their own promise while cold ones proceed untouched.
Key takeaway
Write it in this order: message with topic-stamped sequence, the subscription owning its guarded queue with the drop-oldest-and-count enqueue, the topic whose publish lock covers sequencing plus fan-out (and whose copy-on-write list settles the subscribe race), then the worker loop whose catch block is the delivery contract — re-enqueue for at-least-once, accepted loss for at-most-once. Narrate what every lock covers; that narration is the artifact being graded.