In-Memory Message Streaming Service (Mini Kafka, Machine Coding)

0) Problem Restatement

Flipkart SDE-3 machine coding round: build a working MVP of an in-memory message streaming service, like a tiny Kafka:

• Topics, each split into partitions.
• Producers publish messages (optionally with a key). Messages with the same key go to the same partition.
• Consumers read messages, and consumer groups share work: each partition is read by one consumer of the group.
• Messages keep order within a partition. The system is thread-safe and supports real-time consumption (consumers get new messages as they arrive).

1) Design

classDiagram
    class Broker { +createTopic(name, partitions) +publish(topic, key, value) +subscribe(topic, group, consumerId) }
    class Topic { +name +List partitions }
    class Partition { +List log +lock +condition +append(msg) int +read(offset, max, timeout) List }
    class ConsumerGroup { +name +Map offsets +Map assignment +rebalance(consumers) }
    class Consumer { +id +poll(max, timeout) List +commit() }
    Broker --> Topic
    Topic --> Partition
    Broker --> ConsumerGroup
    ConsumerGroup --> Consumer

• Partition = an append-only list (the log). Each message's offset is its index. Order is guaranteed within a partition.
• Key → partition: hash(key) % num_partitions. No key → round-robin.
• Consumer group stores the committed offset per partition and assigns partitions to its consumers (round-robin). If a consumer joins or leaves, it rebalances.

2) Code (Python)

import threading, itertools

class Partition:
    def __init__(self):
        self.log = []
        self.cond = threading.Condition()
    def append(self, msg):
        with self.cond:
            self.log.append(msg)
            self.cond.notify_all()            # wake consumers waiting for new data
            return len(self.log) - 1
    def read(self, offset, max_n, timeout):
        with self.cond:
            if offset >= len(self.log):
                self.cond.wait(timeout)       # long-poll for real-time consumption
            return self.log[offset:offset + max_n]

class Topic:
    def __init__(self, name, n):
        self.name, self.partitions, self._rr = name, [Partition() for _ in range(n)], itertools.count()
    def pick(self, key):
        n = len(self.partitions)
        return hash(key) % n if key is not None else next(self._rr) % n

class ConsumerGroup:
    def __init__(self, topic):
        self.topic, self.offsets, self.members, self.assign = topic, {}, [], {}
        self.lock = threading.Lock()
    def join(self, cid):
        with self.lock:
            self.members.append(cid); self._rebalance()
    def leave(self, cid):
        with self.lock:
            self.members.remove(cid); self._rebalance()
    def _rebalance(self):
        self.assign = {m: [] for m in self.members}
        for p in range(len(self.topic.partitions)):
            if self.members:
                self.assign[self.members[p % len(self.members)]].append(p)

class Broker:
    def __init__(self):
        self.topics, self.groups, self.lock = {}, {}, threading.Lock()
    def create_topic(self, name, partitions):
        with self.lock: self.topics[name] = Topic(name, partitions)
    def publish(self, topic, value, key=None):
        t = self.topics[topic]; p = t.pick(key)
        return p, t.partitions[p].append((key, value))
    def group(self, topic, name):
        with self.lock:
            return self.groups.setdefault((topic, name), ConsumerGroup(self.topics[topic]))
    def poll(self, topic, group_name, cid, max_n=10, timeout=0.5):
        g = self.group(topic, group_name); out = []
        with g.lock:
            parts = list(g.assign.get(cid, []))
        for p in parts:
            off = g.offsets.get(p, 0)
            msgs = g.topic.partitions[p].read(off, max_n, timeout if not out else 0)
            with g.lock:
                g.offsets[p] = off + len(msgs)   # auto-commit after read (at-most-once style)
            out += [(p, off + i, m) for i, m in enumerate(msgs)]
        return out

3) Explaining the Choices

• Ordering: only within a partition. The key-based partitioning keeps all of one key's messages in order.
• Thread safety: each partition has its own lock (a Condition), so producers to different partitions don't block each other. The group state has its own lock.
• Real-time: read waits on the condition when there's no new data (long-polling), and producers notify_all.
• Delivery semantics: committing after reading = at-most-once. For at-least-once, commit after processing (an explicit commit(partition, offset) call).
• Rebalancing: on join or leave, partitions are redistributed. Committed offsets survive, so a new owner continues where the old one stopped.

4) Extensions (what reviewers ask)

• Retention: delete (or trim) messages older than T, or beyond N per partition, and track a base offset.
• Persistence: an append-only segment file per partition, plus an index of offset → file position.
• Replication (real Kafka): a leader and followers per partition, acknowledged when in-sync replicas have the message.
• Backpressure: bounded partition size, where producers block or get an error when full.

5) Wrap-Up

Model topics as lists of partitions, each an append-only log guarded by its own condition variable. Route messages by key hash (or round-robin), and let consumer groups hold per-partition committed offsets and round-robin partition assignments that rebalance on membership changes. Long-poll reads give real-time consumption with order kept per partition, and retention, persistence and at-least-once commits are natural extensions.