Message Queue / Pub-Sub: a Mini-Kafka in Memory

15.1Problem Statement

Design an in-memory publish-subscribe / message-queue system — a mini-Kafka. Producers publish messages to named topics; consumers subscribe and receive them, either as a broadcast (every subscriber sees every message) or as part of a consumer group (each message goes to exactly one member, load-balanced). Every consumer tracks its own progress and can rewind to replay history.

Functional requirements:

• Topics: named channels. Producers publish messages to a topic.
• Consumers subscribe to a topic and receive messages, either broadcast (every subscriber gets every message) or via a consumer group (each message delivered to exactly one consumer in the group).
• Each consumer tracks its own offset so it resumes correctly, and can replay from an arbitrary offset.
• Delivery must not block producers — publish and consume are decoupled.

Non-functional constraints — read these as the actual grading rubric:

• Thread-safe under many concurrent producers and consumers.
• Observer is the core pattern: subscribers get messages pushed to them, not polled.
• Backpressure / bounded buffers: a slow consumer must not grow memory without bound — block, drop, or cap it, and state the policy.
• Pick at-least-once or at-most-once delivery, and justify the pick.

The expected public interface, given up front so the design effort goes into behavior, not signatures:

using Message = std::string;          // or a struct with payload + metadata

class Subscriber {                     // Observer
public:
    virtual void onMessage(const std::string& topic, const Message& msg) = 0;
    virtual ~Subscriber() = default;
};

class MessageBroker {
public:
    void createTopic(const std::string& topic);
    void publish(const std::string& topic, const Message& msg);                 // non-blocking
    void subscribe(const std::string& topic, std::shared_ptr<Subscriber> sub,
                   const std::string& groupId = "");                            // group => competing consumers
    void resetOffset(const std::string& topic, const std::string& consumerId, size_t offset);  // replay
};

What the interviewer is really testing:

• Observer done at scale, with a producer-consumer queue decoupling publish from delivery.
• Per-consumer offsets and the ability to replay from any of them.
• A concrete concurrency design: per-topic queue, condition variables, a dispatch thread pool.
• A stated backpressure policy for slow consumers — the detail that signals seniority.

Follow-up questions to expect once the base design stands:

• Consumer groups with partition-style load balancing across members.
• At-least-once delivery with acknowledgements and redelivery on timeout.
• Persistence / durability — how the in-memory log extends to disk.
• Ordering guarantees within a partition versus across the whole topic.

15.2How to Approach

Mining the Constraints First

Two sentences in the prompt decide almost the entire architecture. "Delivery should not block producers — decouple publish from consume" means publish() can never call a subscriber inline; it appends and returns, with a buffer sitting between producer and consumer. "Thread-safe under many concurrent producers and consumers" means every shared structure — the topic registry, each topic's log, each consumer's offset — needs a stated locking story; concurrency isn't a follow-up here, it's the body of the answer. A naive Observer (loop over subscribers, call each one from inside publish) satisfies "Observer at the core" but violates the no-blocking rule the instant one subscriber is slow. The twist being tested is Observer plus an asynchronous dispatch layer sitting between the log and the callback.

The Key Insight: the Log Is Not a Queue

The instinctive first draft is a std::queue per topic, with delivery as pop(). That breaks two requirements simultaneously: you can't replay, because the message is already gone, and broadcast is impossible, because the first consumer to pop it consumes it before anyone else sees it.

The fix is to stop treating storage and consumption as the same operation. A topic is an append-only log; each consumer holds an offset, the index of the next entry it should read. Delivery never removes anything — it only advances a cursor.

Topic "orders" log:   [ m0 ][ m1 ][ m2 ][ m3 ][ m4 ]   <- append at tail
                                         ▲              ▲
                          consumerB.offset=2       consumerA.offset=5 (caught up)

Broadcast falls out for free: every consumer walks the whole log independently, so nothing is ever "used up." Replay is one assignment: resetOffset(consumerId, k) moves that consumer's cursor back to k, and the next dispatch re-delivers from there. And bounded memory becomes a question about the slowest reader: the log can't grow past its cap while a lagging cursor still points into the region we'd have to evict.

Consumer groups turn out to need no new structure, just a different cursor owner: instead of one offset per consumer, a group gets one shared offset for the whole group, and members take turns — round-robin — being handed whatever sits at that shared position. Broadcast is a consumer with a private cursor; a group is several consumers sharing one. Same log, different cursor ownership.

Interface as a Threading Contract

The interface sketch above is silent on one thing the design needs: a stable per-consumer key. resetOffset already takes a consumerId, so subscribe has to hand one in too — the interface is quietly implying it even though the first draft didn't spell it out. The bigger point: "publish is non-blocking" is a promise about threading, not about the function signature. Write that promise down before touching internals, so nothing later quietly breaks it — under our backpressure policy publish may still block, but only as deliberate backpressure, never to run a subscriber's callback.

Lock Granularity and the Line User Code Must Never Cross

Each topic gets its own mutex, so traffic on one topic never contends with another — lock granularity is a design decision worth stating out loud, and "one global broker mutex" is the answer that signals you didn't think about it. Two condition variables complete the classic bounded-buffer pair: notFull_, which producers wait on when the log is at capacity, and notEmpty_, the hook a blocking dispatcher would wait on for new work.

Backpressure and Delivery Semantics, Said Out Loud

The prompt explicitly grades the choosing, not just the mechanism. Backpressure: bounded log per topic, block the producer when the slowest reader hasn't drained enough to make room — a slow consumer eventually throttles a fast producer instead of exhausting memory. The stated alternative is drop-oldest: evict the head and shove any lagging cursor forward, bounded but lossy, the right call for telemetry where freshness beats completeness and the wrong one for orders or payments. Delivery: at-most-once as the baseline — a dispatcher commits (advances the offset) right after onMessage returns; if the consumer throws, we skip and move on rather than stall the whole topic on one bad subscriber. The stated upgrade is at-least-once: stop auto-committing, hand out an explicit ack(offset), and redeliver anything that times out un-acked — a deferred commit, not a structural change.

What the Follow-Ups Actually Cost

Every follow-up lands on a seam the design already has, and saying so out loud is the senior move. Consumer groups are a new cursor owner, described above. At-least-once is a deferred commit plus a reaper thread. Persistence is realizing the log already is a write-ahead log — back it with a segment file, fsync before signalling, replay on restart. Ordering is a stated trade: total order within one topic's log, but only per partition once a topic is split for parallelism across group members — the same trade-off Kafka makes, worth naming explicitly rather than discovering under questioning.

15.3Whole Solution

The program below merges every incremental step above into one self-contained C++17 file: the Message/Subscriber leaves, a Topic that owns the append-only log plus both kinds of cursor (per-consumer and per-group), and a MessageBroker facade backed by a small dispatch thread pool. Two refinements beyond the narrative above are worth flagging, because they're exactly the kind of gap synthesis is supposed to close: the dispatch loop needs the real topic name to hand to onMessage (the sketch left that as a placeholder), and a per-topic try_lock around each dispatch attempt stops two dispatcher threads from ever grabbing the same pending message at once — a small addition in the same spirit as the per-topic data mutex already described. main() exercises a standalone broadcast consumer alongside a two-member consumer group on the same topic, then a replay and a rejected replay of an unknown consumer.

// ============================================================
// In-Memory Message Queue / Pub-Sub -- a mini-Kafka
// Observer at scale + producer-consumer concurrency + offsets/backpressure
// Compile: g++ -std=c++17 -O2 -pthread mqueue.cpp -o mqueue
// ============================================================
#include <iostream>
#include <string>
#include <memory>
#include <vector>
#include <unordered_map>
#include <mutex>
#include <condition_variable>
#include <thread>
#include <atomic>
#include <chrono>
#include <stdexcept>
#include <algorithm>
#include <limits>
#include <cstdint>

// ---------- Message: the unit that flows through the system ----------
// id doubles as the message's absolute offset in its topic's log, so
// "offset N" and "message with id N" are the same integer.
struct Message {
    std::string payload;
    uint64_t    id        = 0;
    long long   timestamp = 0;
};

// ---------- Subscriber: the Observer ----------
class Subscriber {
public:
    virtual void onMessage(const std::string& topic, const Message& msg) = 0;
    virtual ~Subscriber() = default;
};

// A trivial concrete observer used by the driver below.
class PrintingSubscriber : public Subscriber {
public:
    explicit PrintingSubscriber(std::string name) : name_(std::move(name)) {}
    void onMessage(const std::string& topic, const Message& msg) override {
        std::cout << "[" << name_ << "] " << topic << "#" << msg.id
                  << " -> " << msg.payload << "\n";
    }
private:
    std::string name_;
};

// ---------- Consumer: a subscriber's registration + its cursor ----------
// groupId == "" means a standalone / broadcast consumer: it owns "offset"
// outright. A non-empty groupId means this consumer competes inside a
// group; "offset" is then unused, because the group's shared cursor
// (Topic::GroupState::offset) is authoritative for every member.
struct Consumer {
    std::string                 id;
    std::string                 groupId;
    std::shared_ptr<Subscriber> sub;
    size_t                      offset = 0;
};

// ---------- Topic: the append-only log + per-consumer/per-group cursors ----------
class Topic {
public:
    // What one dispatch step needs in order to deliver a single message to
    // a single subscriber, without holding any lock while it does so.
    struct Delivery {
        std::string                 consumerId;
        std::string                 groupId;   // "" if this was a standalone/broadcast delivery
        std::shared_ptr<Subscriber> sub;
        Message                     msg;
    };

    explicit Topic(std::string name, size_t capacity = 1024)
        : name_(std::move(name)), capacity_(capacity) {}

    const std::string& name() const { return name_; }

    // Producer side. Blocks if the log is full AND the slowest reader
    // (a standalone consumer or a group) hasn't drained past the point
    // we'd need to evict to make room -- our backpressure policy.
    void append(const Message& msg) {
        std::unique_lock<std::mutex> lk(mtx_);
        notFull_.wait(lk, [&] { return log_.size() - minOffset() < capacity_; });
        Message stamped = msg;
        stamped.id = baseOffset_ + log_.size();     // id == absolute offset
        log_.push_back(std::move(stamped));
        notEmpty_.notify_all();                      // hook for a future blocking dispatcher
    }

    // Register a consumer. Standalone consumers (groupId == "") get their
    // own cursor, starting at the current tail -- new subscribers see only
    // what's published from now on, not ancient history. Group members
    // share one cursor per group: the group's cursor is created at the
    // tail the first time that group is seen, and later joiners simply
    // attach to the existing shared cursor rather than resetting it.
    void addConsumer(const std::string& consumerId, const std::string& groupId,
                      std::shared_ptr<Subscriber> sub) {
        std::lock_guard<std::mutex> lk(mtx_);
        size_t tail = baseOffset_ + log_.size();
        consumers_[consumerId] = Consumer{consumerId, groupId, std::move(sub), tail};
        if (!groupId.empty()) {
            auto it = groups_.find(groupId);
            if (it == groups_.end()) {
                GroupState g;
                g.offset = tail;
                g.members.push_back(consumerId);
                groups_.emplace(groupId, std::move(g));
            } else {
                it->second.members.push_back(consumerId);
            }
        }
    }

    // Replay: rewind a consumer's cursor. For a group member this rewinds
    // the whole group's shared cursor -- a group has exactly one reading
    // position, so replaying one member replays all of them.
    void resetOffset(const std::string& consumerId, size_t offset) {
        std::lock_guard<std::mutex> lk(mtx_);
        auto it = consumers_.find(consumerId);
        if (it == consumers_.end())
            throw std::invalid_argument("unknown consumer: " + consumerId);
        if (it->second.groupId.empty()) {
            it->second.offset = offset;
        } else {
            auto git = groups_.find(it->second.groupId);
            if (git != groups_.end()) git->second.offset = offset;
        }
        notEmpty_.notify_all();
    }

    // Dispatcher side: find the next (consumer, message) pair ready to go,
    // if any. Standalone consumers are checked first, each against its own
    // independent cursor (broadcast: nothing is ever "used up"). Grouped
    // consumers share one cursor per group and are handed messages
    // round-robin among the group's members (partitioned, load-balanced).
    bool nextPending(Delivery& d) {
        std::lock_guard<std::mutex> lk(mtx_);
        for (auto& [id, c] : consumers_) {
            if (!c.groupId.empty()) continue;
            size_t idx = c.offset - baseOffset_;
            if (idx < log_.size()) {
                d.consumerId = id; d.groupId.clear(); d.sub = c.sub; d.msg = log_[idx];
                return true;
            }
        }
        for (auto& [gid, g] : groups_) {
            if (g.members.empty()) continue;
            size_t idx = g.offset - baseOffset_;
            if (idx >= log_.size()) continue;
            const std::string& memberId = g.members[g.rrIndex % g.members.size()];
            auto mit = consumers_.find(memberId);
            if (mit == consumers_.end()) continue;
            d.consumerId = memberId; d.groupId = gid; d.sub = mit->second.sub; d.msg = log_[idx];
            return true;
        }
        return false;
    }

    // Called by the dispatcher AFTER a successful onMessage -- our
    // at-most-once commit. Advances the right cursor (the consumer's own,
    // or the group's shared one) and rotates the group's round-robin.
    void commit(const std::string& consumerId, const std::string& groupId) {
        std::lock_guard<std::mutex> lk(mtx_);
        if (groupId.empty()) {
            auto it = consumers_.find(consumerId);
            if (it != consumers_.end()) ++it->second.offset;
        } else {
            auto git = groups_.find(groupId);
            if (git != groups_.end()) {
                ++git->second.offset;
                git->second.rrIndex =
                    (git->second.rrIndex + 1) % std::max<size_t>(1, git->second.members.size());
            }
        }
        notFull_.notify_all();                        // a reader advanced -> maybe room now
    }

    // Serializes one topic's dispatch attempts so two dispatcher threads
    // can never grab the same pending message for the same consumer or
    // group at once. Held only around fetch+commit, never across the
    // subscriber callback's own work -- nextPending/commit copy out what
    // they need and return quickly (see MessageBroker::dispatchLoop).
    std::mutex& dispatchMutex() { return dispatchMtx_; }

private:
    // Smallest offset any reader (a standalone consumer or a group) still
    // needs -- the floor below which the log may not grow past capacity_
    // without blocking the producer (our backpressure policy).
    size_t minOffset() const {
        if (consumers_.empty() && groups_.empty()) return baseOffset_ + log_.size();
        size_t m = std::numeric_limits<size_t>::max();
        for (auto& [id, c] : consumers_) if (c.groupId.empty()) m = std::min(m, c.offset);
        for (auto& [gid, g] : groups_) m = std::min(m, g.offset);
        return m;
    }

    struct GroupState {
        size_t                   offset  = 0;   // the group's one shared cursor
        std::vector<std::string> members;       // round-robin order
        size_t                   rrIndex = 0;   // index into members for the next delivery
    };

    std::string name_;
    size_t      capacity_;
    size_t      baseOffset_ = 0;                // offset of log_[0]; this design never evicts,
                                                 // so baseOffset_ stays 0 (see 15.2 on backpressure)
    std::vector<Message> log_;
    std::unordered_map<std::string, Consumer>   consumers_;
    std::unordered_map<std::string, GroupState> groups_;
    std::mutex               mtx_;              // guards log_, consumers_, groups_
    std::mutex               dispatchMtx_;      // serializes dispatch attempts per topic
    std::condition_variable  notEmpty_;         // reserved for a future CV-based dispatcher
    std::condition_variable  notFull_;          // producers wait here -- backpressure
};

// ---------- MessageBroker: facade + dispatch thread pool ----------
class MessageBroker {
public:
    explicit MessageBroker(unsigned dispatchers = std::thread::hardware_concurrency()) {
        running_ = true;
        for (unsigned i = 0; i < std::max(1u, dispatchers); ++i)
            pool_.emplace_back([this] { dispatchLoop(); });
    }

    ~MessageBroker() {
        running_ = false;             // dispatchLoop polls this every spin; no wake-up needed
        for (auto& th : pool_) if (th.joinable()) th.join();
    }

    void createTopic(const std::string& topic) {
        std::lock_guard<std::mutex> lk(reg_mtx_);
        topics_.emplace(topic, std::make_unique<Topic>(topic));
    }

    // Non-blocking, except for the deliberate backpressure block inside
    // Topic::append() when the log is at capacity.
    void publish(const std::string& topic, const Message& msg) {
        find(topic)->append(msg);
    }

    void subscribe(const std::string& topic, std::shared_ptr<Subscriber> sub,
                   const std::string& consumerId, const std::string& groupId = "") {
        find(topic)->addConsumer(consumerId, groupId, std::move(sub));
    }

    void resetOffset(const std::string& topic, const std::string& consumerId, size_t offset) {
        find(topic)->resetOffset(consumerId, offset);
    }

private:
    Topic* find(const std::string& topic) {
        std::lock_guard<std::mutex> lk(reg_mtx_);
        auto it = topics_.find(topic);
        if (it == topics_.end()) throw std::invalid_argument("no such topic: " + topic);
        return it->second.get();
    }

    // Each dispatcher thread repeatedly: try every topic, and if no other
    // dispatcher currently owns it, pull one pending delivery, call
    // onMessage with NO lock held, then commit.
    void dispatchLoop() {
        while (running_) {
            bool didWork = false;
            std::vector<Topic*> snapshot;
            { std::lock_guard<std::mutex> lk(reg_mtx_);
              for (auto& [n, t] : topics_) snapshot.push_back(t.get()); }

            for (Topic* t : snapshot) {
                std::unique_lock<std::mutex> dlk(t->dispatchMutex(), std::try_to_lock);
                if (!dlk.owns_lock()) continue;        // another thread already owns this topic

                Topic::Delivery d;
                if (t->nextPending(d)) {
                    try { d.sub->onMessage(t->name(), d.msg); }
                    catch (...) { /* at-most-once: log and move on */ }
                    t->commit(d.consumerId, d.groupId);
                    didWork = true;
                }
            }
            if (!didWork) std::this_thread::yield();   // production: wait on a CV instead of spin
        }
    }

    std::unordered_map<std::string, std::unique_ptr<Topic>> topics_;
    std::mutex                reg_mtx_;                 // guards the topic registry only
    std::vector<std::thread>  pool_;
    std::atomic<bool>         running_{false};
};

// ---------- Driver: broadcast + consumer group + replay + error path ----------
int main() {
    MessageBroker broker(/*dispatchers=*/3);
    broker.createTopic("orders");

    auto alice = std::make_shared<PrintingSubscriber>("alice");
    auto w1    = std::make_shared<PrintingSubscriber>("worker-1");
    auto w2    = std::make_shared<PrintingSubscriber>("worker-2");

    // Broadcast: alice is a standalone consumer with her own cursor --
    // she will see every message published to "orders".
    broker.subscribe("orders", alice, "alice");

    // Consumer group "shippers": worker-1 and worker-2 share one cursor
    // and split the stream between them, round-robin.
    broker.subscribe("orders", w1, "worker-1", "shippers");
    broker.subscribe("orders", w2, "worker-2", "shippers");

    for (int i = 0; i < 6; ++i)
        broker.publish("orders", Message{"order-" + std::to_string(i)});

    std::this_thread::sleep_for(std::chrono::milliseconds(150));  // let dispatch drain

    // Replay: rewind alice to offset 0; she re-receives the whole log.
    // The group's shared cursor is untouched -- workers do not replay.
    std::cout << "--- alice replays from 0 ---\n";
    broker.resetOffset("orders", "alice", 0);
    std::this_thread::sleep_for(std::chrono::milliseconds(150));

    // resetOffset on an unknown consumer is rejected, not crashed.
    try { broker.resetOffset("orders", "ghost", 0); }
    catch (const std::exception& e) { std::cout << "Rejected: " << e.what() << "\n"; }

    return 0;   // ~MessageBroker stops and joins the dispatch pool cleanly
}
// Expected shape of output: "order-0".."order-5" printed once each for
// alice (broadcast); the same six messages split between worker-1 and
// worker-2 (round-robin within "shippers"), interleaved with alice's in
// some order; then all six reprinted for alice only (replay); then the
// "Rejected: unknown consumer: ghost" line.