Machine coding · Schedulers and queues

Topic-based publish-subscribe with consumer groups

Reported once
SDE-2
Observer
Strategy
Publish-subscribe
1 source

One candidate write-up. Treat it as a single data point.

The prompt

Build a system where many senders publish events to named topics and many receivers subscribe to topics, so that every event on a topic reaches every subscriber. As a bonus, support receiver groups where each group gets every event once and only one member of the group handles it, in the way Kafka consumer groups behave.

Restated in our own words, not a quote from Ola or any page. The answer below is written for this site.

Try it first.The answer is covered until you pick. Talk it through against a 60 minute clock, or just read it.

Clarify first

Pin the scope down before you name a single class.

The statement is short, so confirm these before coding.

Does a new subscriber see past events, or only events published after it subscribed? Assume only later events.
Is delivery push (the broker calls a handler) or pull (the consumer polls)? Assume push into a per-consumer inbox that a delivery step drains.
What ordering is needed? Assume events reach each consumer in publish order.
Inside a group, how is the member chosen? Assume a replaceable rule, with round robin by default and a key hash as the second option.
What happens when a handler fails? Assume the failure is recorded and does not stop other consumers.

In scope: topics, publishers, subscribers, groups, members joining and leaving. Out of scope: persistence, replay from an offset, acknowledgements with redelivery, and partitions.

Entities and relationships

Who owns what, and who talks to whom.

Broker is the facade. It owns the topics and the dead letter list, and offers createTopic, subscribe, joinGroup, leaveGroup, publish and deliver.
Topic has a sequence counter, a map of plain subscribers and a map of groups. Publishing stamps an event and fans it out.
Publisher is a thin sender with an id that calls the broker, so the sender is recorded on each event.
Endpoint is a receiver: an id, a handler function and an inbox queue. It is used both for plain subscribers and for group members.
Group has many endpoints, a backlog for events that arrive when it has no members, and a member assignment strategy.
RoundRobin and KeyHash implement the strategy with one method, pick(members, event).

The design

The class map, the main flow and the patterns that earn their place.

Pub-sub class mapA topic copies each event to plain subscribers and offers it once to each group, which picks one member.

A publisher sends a payload to a named topic through the broker.
The topic stamps the next sequence number and builds the event.
Every plain subscriber gets the event in its inbox.
Every group is offered the event exactly once.
The group asks its strategy which member should handle it.
The event lands in that member's inbox only.
On delivery, a handler that throws is recorded and the rest carry on.

Open on the whiteboardAll 2 diagrams on one tidy board, with a step-by-step walkthrough.

Publishing has two fan-out rules on the same topic. A plain subscriber is an Observer: it gets every event, so the topic copies the event reference into each subscriber's inbox. A group is one logical subscriber: the topic offers the event to the group once, and the group's assignment strategy chooses one member whose inbox receives it. Round robin spreads load evenly; key hash sends events with the same key to the same member, which keeps per-key order. Both sit behind pick, so adding sticky or least-loaded assignment is a new class.

Main flow:

publish(topic, payload, key) looks up the topic (unknown topics throw), increments its sequence number and builds the event.
The event goes into the inbox of every plain subscriber and is offered to every group.
A group with members calls the strategy and pushes the event into that member's inbox. A group with no members keeps the event in its backlog, and the next member to join receives it.
deliver walks every endpoint, drains its inbox in order and calls the handler. A handler that throws is recorded in the dead letter list with the endpoint id and sequence number, and delivery carries on. The event is not retried.

Each consumer has its own inbox, so one slow or failing consumer cannot hold up the others, and each consumer sees events in publish order. When a member leaves, its undelivered inbox is re-offered to the group and merged into the receiving inbox by sequence number, so nothing is lost and order still holds. A plain unsubscribe discards that subscriber's undelivered events, and the first member to join a group fixes its strategy. Events already sitting in other inboxes stay where they are when someone joins, so a key can briefly be split across two members during a membership change. Consistent hashing reduces that.

A real version needs one lock per topic so sequence numbers and fan-out happen atomically, a concurrent or copy-on-write subscriber list so subscribe and unsubscribe do not break a publish in progress, and a worker thread or executor per endpoint that blocks on its inbox. Handlers must run outside the topic lock.

Trade-off: inboxes per consumer are simple and isolate failures, but memory grows with slow consumers and a crash loses undelivered events. The alternative is a shared append-only log per topic with an offset per subscriber or group. That gives replay, late subscribers reading from the start and cheap fan-out, at the cost of retention and offset management.

The code

Runnable JavaScript, with the checks it passes.

ola-pubsub-topics.js

function hash(value) {
  let h = 0;
  for (const ch of String(value)) h = (h * 31 + ch.charCodeAt(0)) >>> 0;
  return h;
}

class RoundRobin {
  constructor() {
    this.next = 0;
  }
  pick(members) {
    const member = members[this.next % members.length];
    this.next += 1;
    return member;
  }
}

class KeyHash {
  pick(members, event) {
    return members[hash(event.key ?? event.seq) % members.length];
  }
}

class Endpoint {
  constructor(id, handler) {
    this.id = id;
    this.handler = handler;
    this.inbox = [];
  }
}

class Group {
  constructor(id, strategy) {
    this.id = id;
    this.strategy = strategy;
    this.members = [];
    this.backlog = [];
  }
  accept(event) {
    if (this.members.length === 0) {
      this.backlog.push(event);
      return;
    }
    const inbox = this.strategy.pick(this.members, event).inbox;
    inbox.push(event);
    if (inbox.length > 1 && inbox[inbox.length - 2].seq > event.seq) inbox.sort((a, b) => a.seq - b.seq);
  }
  join(endpoint) {
    if (this.members.some((member) => member.id === endpoint.id)) throw new Error("already in group: " + endpoint.id);
    this.members.push(endpoint);
    const waiting = this.backlog.splice(0);
    for (const event of waiting) this.accept(event);
  }
  leave(id) {
    const index = this.members.findIndex((member) => member.id === id);
    if (index < 0) return false;
    const [gone] = this.members.splice(index, 1);
    for (const event of gone.inbox.splice(0)) this.accept(event);
    return true;
  }
}

class Topic {
  constructor(name) {
    this.name = name;
    this.subscribers = new Map();
    this.groups = new Map();
    this.seq = 0;
  }
  publish(payload, key, sender) {
    this.seq += 1;
    const event = { topic: this.name, seq: this.seq, key, payload, sender };
    for (const endpoint of this.subscribers.values()) endpoint.inbox.push(event);
    for (const group of this.groups.values()) group.accept(event);
    return event.seq;
  }
}

class Broker {
  constructor() {
    this.topics = new Map();
    this.deadLetters = [];
  }
  createTopic(name) {
    if (this.topics.has(name)) throw new Error("topic exists: " + name);
    this.topics.set(name, new Topic(name));
  }
  topic(name) {
    const topic = this.topics.get(name);
    if (!topic) throw new Error("unknown topic " + name);
    return topic;
  }
  subscribe(topicName, id, handler) {
    const topic = this.topic(topicName);
    if (topic.subscribers.has(id)) throw new Error("already subscribed: " + id);
    topic.subscribers.set(id, new Endpoint(id, handler));
  }
  unsubscribe(topicName, id) {
    return this.topic(topicName).subscribers.delete(id);
  }
  joinGroup(topicName, groupId, memberId, handler, strategy) {
    const topic = this.topic(topicName);
    if (!topic.groups.has(groupId)) topic.groups.set(groupId, new Group(groupId, strategy ?? new RoundRobin()));
    topic.groups.get(groupId).join(new Endpoint(memberId, handler));
  }
  leaveGroup(topicName, groupId, memberId) {
    const group = this.topic(topicName).groups.get(groupId);
    return group ? group.leave(memberId) : false;
  }
  publish(topicName, payload, key, sender) {
    return this.topic(topicName).publish(payload, key, sender);
  }
  deliver() {
    let handled = 0;
    for (const topic of this.topics.values()) {
      const endpoints = [...topic.subscribers.values()];
      for (const group of topic.groups.values()) endpoints.push(...group.members);
      for (const endpoint of endpoints) {
        for (const event of endpoint.inbox.splice(0)) {
          try {
            endpoint.handler(event);
            handled += 1;
          } catch (error) {
            const message = error instanceof Error ? error.message : String(error);
            this.deadLetters.push({ endpoint: endpoint.id, seq: event.seq, error: message });
          }
        }
      }
    }
    return handled;
  }
}

class Publisher {
  constructor(id, broker) {
    this.id = id;
    this.broker = broker;
  }
  send(topic, payload, key) {
    return this.broker.publish(topic, payload, key, this.id);
  }
}

Runs here in your browser, in a sandboxed worker. Nothing is sent anywhere.

Try your own input

Write JavaScript that calls the code above and prints with console.log. It runs after the solution.

Extend it

The change requests an interviewer throws next.

A new subscriber should be able to read the topic from the beginning. What storage and bookkeeping does that need, and what does it do to memory?
A handler fails. How would you add acknowledgements, retries and a dead letter queue so that a group member's failure does not lose the event?
How would you guarantee that events with the same key are handled in order by one member even when members join and leave?
Publish and subscribe now happen on many threads at once. Which operations need a lock, and which data structures can be lock-free?
A topic gets far more traffic than one consumer can handle. How would you split it into partitions and how would a group share them?

Where candidates slip

The mistakes this problem punishes.

Delivering by calling handlers inside publish. A slow or throwing subscriber then blocks the publisher and every subscriber after it.
Sharing one inbox per topic and letting subscribers read from it with a counter, which makes removing events and group handling awkward.
Giving a group event to every member, or to none when the picked member has left. Exactly one member must get each event.
Forgetting that a group is separate from plain subscribers: two groups on one topic each receive every event once.
Losing a leaving member's pending events instead of re-assigning them.
Using wall-clock time or random numbers for ordering and assignment, which makes the demo non-repeatable. Use a sequence number and a stable hash.

Have a go before you read.

Interviews reward the sketch you make cold. Clarify out loud, draw it, then compare with the answer.

Always show the answer straight away

Now make it stick

Sketch the class map on the whiteboard Run a mock interview

Where this was reported

Public pages only. Companies do not confirm these, and they rotate their questions, so treat each link as one person's account or one list's tag.

Candidate write-upOLA Interview Experience, Set 8 for SDE-2 (GeeksforGeeks)

Read these chapters first

Message queues & async processing

Now make it stick

Where this was reported

Read these chapters first

More on schedulers and queues