Designing a Pub/Sub Messaging System (Mini-Kafka)

import time


class Message:
    def __init__(self, offset: int, key: str | None, value: str):
        self._offset = offset
        self._key = key
        self._value = value
        self._timestamp = time.time()

    @property
    def offset(self) -> int:
        return self._offset

    @property
    def key(self) -> str | None:
        return self._key

    @property
    def value(self) -> str:
        return self._value

    @property
    def timestamp(self) -> float:
        return self._timestamp

    def __repr__(self) -> str:
        return f"Message(offset={self._offset}, key='{self._key}', value='{self._value}')"

#pragma once
#include <string>
#include <chrono>

class Message {
private:
    int offset_;
    std::string key_;
    std::string value_;
    long long timestamp_;

public:
    Message(int offset, std::string key, std::string value)
        : offset_(offset), key_(std::move(key)), value_(std::move(value)) {
        timestamp_ = std::chrono::duration_cast<std::chrono::milliseconds>(
            std::chrono::system_clock::now().time_since_epoch()).count();
    }

    int getOffset() const { return offset_; }
    const std::string& getKey() const { return key_; }
    const std::string& getValue() const { return value_; }
    long long getTimestamp() const { return timestamp_; }

    std::string toString() const {
        return "Message{offset=" + std::to_string(offset_) +
               ", key='" + key_ + "', value='" + value_ + "'}";
    }
};

import threading
from typing import List


class Partition:
    def __init__(self, partition_id: int, topic_name: str):
        self._id = partition_id
        self._topic_name = topic_name
        self._messages: List[Message] = []
        self._lock = threading.RWLock() if hasattr(threading, 'RWLock') else threading.Lock()

    def append(self, key: str | None, value: str) -> int:
        """Append a message. Returns the assigned offset."""
        with self._lock:
            offset = len(self._messages)
            msg = Message(offset, key, value)
            self._messages.append(msg)
            return offset

    def read(self, from_offset: int, max_count: int) -> List[Message]:
        """Read up to max_count messages starting from from_offset."""
        with self._lock:
            if from_offset >= len(self._messages):
                return []
            to_index = min(from_offset + max_count, len(self._messages))
            return list(self._messages[from_offset:to_index])

    @property
    def latest_offset(self) -> int:
        with self._lock:
            return len(self._messages)

    @property
    def id(self) -> int:
        return self._id

    @property
    def topic_name(self) -> str:
        return self._topic_name

    def __repr__(self) -> str:
        return f"{self._topic_name}-partition-{self._id}"

    def __hash__(self) -> int:
        return hash((self._topic_name, self._id))

    def __eq__(self, other) -> bool:
        if not isinstance(other, Partition):
            return False
        return self._id == other._id and self._topic_name == other._topic_name

#pragma once
#include <vector>
#include <string>
#include <shared_mutex>
#include <algorithm>
#include "Message.hpp"

class Partition {
private:
    int id_;
    std::string topicName_;
    std::vector<Message> messages_;
    mutable std::shared_mutex mutex_;

public:
    Partition(int id, std::string topicName)
        : id_(id), topicName_(std::move(topicName)) {}

    // Non-copyable due to mutex
    Partition(const Partition&) = delete;
    Partition& operator=(const Partition&) = delete;

    int append(const std::string& key, const std::string& value) {
        std::unique_lock lock(mutex_);
        int offset = static_cast<int>(messages_.size());
        messages_.emplace_back(offset, key, value);
        return offset;
    }

    std::vector<Message> read(int fromOffset, int maxCount) const {
        std::shared_lock lock(mutex_);
        if (fromOffset >= static_cast<int>(messages_.size())) {
            return {};
        }
        int toIndex = std::min(fromOffset + maxCount,
                               static_cast<int>(messages_.size()));
        return std::vector<Message>(
            messages_.begin() + fromOffset,
            messages_.begin() + toIndex);
    }

    int getLatestOffset() const {
        std::shared_lock lock(mutex_);
        return static_cast<int>(messages_.size());
    }

    int getId() const { return id_; }
    const std::string& getTopicName() const { return topicName_; }

    bool operator==(const Partition& other) const {
        return id_ == other.id_ && topicName_ == other.topicName_;
    }
};

import itertools
from typing import List


class Topic:
    def __init__(self, name: str, partition_count: int):
        if partition_count <= 0:
            raise ValueError("Partition count must be positive")
        self._name = name
        self._partitions = [Partition(i, name) for i in range(partition_count)]
        self._round_robin = itertools.cycle(range(partition_count))

    def publish(self, key: str | None, value: str) -> None:
        """
        Publish a message. Key-based hash for partition routing,
        or round-robin if key is None.
        """
        if key is not None:
            partition_index = abs(hash(key)) % len(self._partitions)
        else:
            partition_index = next(self._round_robin)
        self._partitions[partition_index].append(key, value)

    def get_partition(self, partition_id: int) -> Partition:
        if partition_id < 0 or partition_id >= len(self._partitions):
            raise ValueError(f"Invalid partition id: {partition_id}")
        return self._partitions[partition_id]

    @property
    def partitions(self) -> List[Partition]:
        return list(self._partitions)

    @property
    def partition_count(self) -> int:
        return len(self._partitions)

    @property
    def name(self) -> str:
        return self._name

    def __repr__(self) -> str:
        return f"Topic(name='{self._name}', partitions={len(self._partitions)})"

#pragma once
#include <vector>
#include <string>
#include <memory>
#include <atomic>
#include <cmath>
#include <stdexcept>
#include <functional>
#include "Partition.hpp"

class Topic {
private:
    std::string name_;
    std::vector<std::unique_ptr<Partition>> partitions_;
    std::atomic<int> roundRobinCounter_{0};

public:
    Topic(std::string name, int partitionCount)
        : name_(std::move(name)) {
        if (partitionCount <= 0) {
            throw std::invalid_argument("Partition count must be positive");
        }
        for (int i = 0; i < partitionCount; i++) {
            partitions_.push_back(std::make_unique<Partition>(i, name_));
        }
    }

    void publish(const std::string& key, const std::string& value) {
        int partitionIndex;
        if (!key.empty()) {
            partitionIndex = std::abs(static_cast<int>(
                std::hash<std::string>{}(key) % partitions_.size()));
        } else {
            partitionIndex = std::abs(
                roundRobinCounter_.fetch_add(1) % static_cast<int>(partitions_.size()));
        }
        partitions_[partitionIndex]->append(key, value);
    }

    Partition* getPartition(int id) {
        if (id < 0 || id >= static_cast<int>(partitions_.size())) {
            throw std::invalid_argument("Invalid partition id");
        }
        return partitions_[id].get();
    }

    int getPartitionCount() const {
        return static_cast<int>(partitions_.size());
    }

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

    // Get raw pointers to partitions for iteration
    std::vector<Partition*> getPartitions() {
        std::vector<Partition*> result;
        for (auto& p : partitions_) {
            result.push_back(p.get());
        }
        return result;
    }
};

from typing import List, Dict


class Consumer:
    def __init__(self, consumer_id: str, group_id: str):
        self._id = consumer_id
        self._group_id = group_id
        self._local_offsets: Dict[Partition, int] = {}

    def poll(self, partition: Partition, max_messages: int) -> List[Message]:
        """
        Poll messages from a partition starting at current local offset.
        Does NOT commit - call commit_offset() after processing.
        """
        current_offset = self._local_offsets.get(partition, 0)
        messages = partition.read(current_offset, max_messages)
        if messages:
            self._local_offsets[partition] = current_offset + len(messages)
        return messages

    def commit_offset(self, partition: Partition, offset: int) -> None:
        """Commit offset after successful processing."""
        self._local_offsets[partition] = offset

    def seek(self, partition: Partition, offset: int) -> None:
        """Reset offset to a specific position for replay."""
        self._local_offsets[partition] = offset

    def get_offset(self, partition: Partition) -> int:
        return self._local_offsets.get(partition, 0)

    @property
    def id(self) -> str:
        return self._id

    @property
    def group_id(self) -> str:
        return self._group_id

    def __repr__(self) -> str:
        return f"Consumer(id='{self._id}', group='{self._group_id}')"

    def __eq__(self, other) -> bool:
        if not isinstance(other, Consumer):
            return False
        return self._id == other._id

    def __hash__(self) -> int:
        return hash(self._id)

#pragma once
#include <string>
#include <unordered_map>
#include <vector>
#include "Partition.hpp"
#include "Message.hpp"

class Consumer {
private:
    std::string id_;
    std::string groupId_;
    std::unordered_map<Partition*, int> localOffsets_;

public:
    Consumer(std::string id, std::string groupId)
        : id_(std::move(id)), groupId_(std::move(groupId)) {}

    std::vector<Message> poll(Partition* partition, int maxMessages) {
        int currentOffset = 0;
        auto it = localOffsets_.find(partition);
        if (it != localOffsets_.end()) {
            currentOffset = it->second;
        }
        auto messages = partition->read(currentOffset, maxMessages);
        if (!messages.empty()) {
            localOffsets_[partition] = currentOffset + static_cast<int>(messages.size());
        }
        return messages;
    }

    void commitOffset(Partition* partition, int offset) {
        localOffsets_[partition] = offset;
    }

    void seek(Partition* partition, int offset) {
        localOffsets_[partition] = offset;
    }

    int getOffset(Partition* partition) const {
        auto it = localOffsets_.find(partition);
        return it != localOffsets_.end() ? it->second : 0;
    }

    const std::string& getId() const { return id_; }
    const std::string& getGroupId() const { return groupId_; }

    bool operator==(const Consumer& other) const {
        return id_ == other.id_;
    }
};

import threading
from typing import List, Dict, Optional


class ConsumerGroup:
    def __init__(self, group_id: str):
        self._group_id = group_id
        self._consumers: List[Consumer] = []
        self._partition_assignment: Dict[Partition, Consumer] = {}
        self._committed_offsets: Dict[Partition, int] = {}
        self._subscribed_partitions: List[Partition] = []
        self._lock = threading.Lock()

    def add_consumer(self, consumer: Consumer) -> None:
        with self._lock:
            self._consumers.append(consumer)
            self._rebalance()

    def remove_consumer(self, consumer: Consumer) -> None:
        with self._lock:
            self._consumers.remove(consumer)
            self._rebalance()

    def subscribe(self, partitions: List[Partition]) -> None:
        with self._lock:
            self._subscribed_partitions = list(partitions)
            self._rebalance()

    def _rebalance(self) -> None:
        """Round-robin partition assignment across consumers."""
        self._partition_assignment.clear()
        if not self._consumers or not self._subscribed_partitions:
            return
        for i, partition in enumerate(self._subscribed_partitions):
            consumer = self._consumers[i % len(self._consumers)]
            self._partition_assignment[partition] = consumer

    def commit_offset(self, partition: Partition, offset: int) -> None:
        self._committed_offsets[partition] = offset

    def get_committed_offset(self, partition: Partition) -> int:
        return self._committed_offsets.get(partition, 0)

    def get_assigned_consumer(self, partition: Partition) -> Optional[Consumer]:
        return self._partition_assignment.get(partition)

    def get_assigned_partitions(self, consumer: Consumer) -> List[Partition]:
        return [p for p, c in self._partition_assignment.items() if c == consumer]

    @property
    def group_id(self) -> str:
        return self._group_id

    @property
    def consumers(self) -> List[Consumer]:
        return list(self._consumers)

    def __repr__(self) -> str:
        return (f"ConsumerGroup(id='{self._group_id}', "
                f"consumers={len(self._consumers)}, "
                f"partitions={len(self._subscribed_partitions)})")

#pragma once
#include <string>
#include <vector>
#include <unordered_map>
#include <mutex>
#include <memory>
#include "Consumer.hpp"
#include "Partition.hpp"

class ConsumerGroup {
private:
    std::string groupId_;
    std::vector<Consumer*> consumers_;
    std::unordered_map<Partition*, Consumer*> partitionAssignment_;
    std::unordered_map<Partition*, int> committedOffsets_;
    std::vector<Partition*> subscribedPartitions_;
    mutable std::mutex mutex_;

    void rebalance() {
        partitionAssignment_.clear();
        if (consumers_.empty() || subscribedPartitions_.empty()) return;
        for (size_t i = 0; i < subscribedPartitions_.size(); i++) {
            Consumer* assigned = consumers_[i % consumers_.size()];
            partitionAssignment_[subscribedPartitions_[i]] = assigned;
        }
    }

public:
    ConsumerGroup(std::string groupId)
        : groupId_(std::move(groupId)) {}

    void addConsumer(Consumer* consumer) {
        std::lock_guard lock(mutex_);
        consumers_.push_back(consumer);
        rebalance();
    }

    void removeConsumer(Consumer* consumer) {
        std::lock_guard lock(mutex_);
        consumers_.erase(
            std::remove(consumers_.begin(), consumers_.end(), consumer),
            consumers_.end());
        rebalance();
    }

    void subscribe(const std::vector<Partition*>& partitions) {
        std::lock_guard lock(mutex_);
        subscribedPartitions_ = partitions;
        rebalance();
    }

    void commitOffset(Partition* partition, int offset) {
        std::lock_guard lock(mutex_);
        committedOffsets_[partition] = offset;
    }

    int getCommittedOffset(Partition* partition) const {
        std::lock_guard lock(mutex_);
        auto it = committedOffsets_.find(partition);
        return it != committedOffsets_.end() ? it->second : 0;
    }

    Consumer* getAssignedConsumer(Partition* partition) const {
        std::lock_guard lock(mutex_);
        auto it = partitionAssignment_.find(partition);
        return it != partitionAssignment_.end() ? it->second : nullptr;
    }

    std::vector<Partition*> getAssignedPartitions(Consumer* consumer) const {
        std::lock_guard lock(mutex_);
        std::vector<Partition*> result;
        for (const auto& [partition, assigned] : partitionAssignment_) {
            if (assigned == consumer) {
                result.push_back(partition);
            }
        }
        return result;
    }

    const std::string& getGroupId() const { return groupId_; }
};

class Producer:
    def __init__(self, producer_id: str):
        self._id = producer_id

    def publish(self, topic: Topic, key: str | None, value: str) -> None:
        """Publish a message to the given topic."""
        if topic is None:
            raise ValueError("Topic cannot be None")
        if value is None:
            raise ValueError("Value cannot be None")
        topic.publish(key, value)

    @property
    def id(self) -> str:
        return self._id

    def __repr__(self) -> str:
        return f"Producer(id='{self._id}')"

#pragma once
#include <string>
#include <stdexcept>
#include "Topic.hpp"

class Producer {
private:
    std::string id_;

public:
    Producer(std::string id) : id_(std::move(id)) {}

    void publish(Topic* topic, const std::string& key, const std::string& value) {
        if (!topic) {
            throw std::invalid_argument("Topic cannot be null");
        }
        if (value.empty()) {
            throw std::invalid_argument("Value cannot be empty");
        }
        topic->publish(key, value);
    }

    const std::string& getId() const { return id_; }
};

import threading
from typing import Dict


class PubSubBroker:
    def __init__(self):
        self._topics: Dict[str, Topic] = {}
        self._consumer_groups: Dict[str, ConsumerGroup] = {}
        self._lock = threading.Lock()

    def create_topic(self, name: str, partition_count: int) -> Topic:
        """Create a new topic with the specified number of partitions."""
        with self._lock:
            if name in self._topics:
                raise ValueError(f"Topic '{name}' already exists")
            topic = Topic(name, partition_count)
            self._topics[name] = topic
            return topic

    def get_topic(self, name: str) -> Topic:
        if name not in self._topics:
            raise ValueError(f"Topic '{name}' not found")
        return self._topics[name]

    def create_consumer_group(self, group_id: str) -> ConsumerGroup:
        """Create a consumer group."""
        with self._lock:
            if group_id in self._consumer_groups:
                raise ValueError(f"Group '{group_id}' already exists")
            group = ConsumerGroup(group_id)
            self._consumer_groups[group_id] = group
            return group

    def get_consumer_group(self, group_id: str) -> ConsumerGroup:
        if group_id not in self._consumer_groups:
            raise ValueError(f"Group '{group_id}' not found")
        return self._consumer_groups[group_id]

    def subscribe(self, group_id: str, topic_name: str) -> None:
        """Subscribe a consumer group to a topic."""
        group = self.get_consumer_group(group_id)
        topic = self.get_topic(topic_name)
        group.subscribe(topic.partitions)

    def register_consumer(self, group_id: str, consumer: Consumer) -> None:
        """Register a consumer into a group. Triggers rebalance."""
        group = self.get_consumer_group(group_id)
        group.add_consumer(consumer)

    def deregister_consumer(self, group_id: str, consumer: Consumer) -> None:
        """Deregister a consumer from its group. Triggers rebalance."""
        group = self.get_consumer_group(group_id)
        group.remove_consumer(consumer)

    def has_topic(self, name: str) -> bool:
        return name in self._topics

    def __repr__(self) -> str:
        return (f"PubSubBroker(topics={len(self._topics)}, "
                f"groups={len(self._consumer_groups)})")

#pragma once
#include <string>
#include <unordered_map>
#include <memory>
#include <mutex>
#include <stdexcept>
#include "Topic.hpp"
#include "ConsumerGroup.hpp"
#include "Consumer.hpp"

class PubSubBroker {
private:
    std::unordered_map<std::string, std::unique_ptr<Topic>> topics_;
    std::unordered_map<std::string, std::unique_ptr<ConsumerGroup>> consumerGroups_;
    mutable std::mutex mutex_;

public:
    Topic* createTopic(const std::string& name, int partitionCount) {
        std::lock_guard lock(mutex_);
        if (topics_.count(name)) {
            throw std::invalid_argument("Topic '" + name + "' already exists");
        }
        topics_[name] = std::make_unique<Topic>(name, partitionCount);
        return topics_[name].get();
    }

    Topic* getTopic(const std::string& name) {
        auto it = topics_.find(name);
        if (it == topics_.end()) {
            throw std::invalid_argument("Topic '" + name + "' not found");
        }
        return it->second.get();
    }

    ConsumerGroup* createConsumerGroup(const std::string& groupId) {
        std::lock_guard lock(mutex_);
        if (consumerGroups_.count(groupId)) {
            throw std::invalid_argument("Group '" + groupId + "' already exists");
        }
        consumerGroups_[groupId] = std::make_unique<ConsumerGroup>(groupId);
        return consumerGroups_[groupId].get();
    }

    ConsumerGroup* getConsumerGroup(const std::string& groupId) {
        auto it = consumerGroups_.find(groupId);
        if (it == consumerGroups_.end()) {
            throw std::invalid_argument("Group '" + groupId + "' not found");
        }
        return it->second.get();
    }

    void subscribe(const std::string& groupId, const std::string& topicName) {
        auto* group = getConsumerGroup(groupId);
        auto* topic = getTopic(topicName);
        group->subscribe(topic->getPartitions());
    }

    void registerConsumer(const std::string& groupId, Consumer* consumer) {
        auto* group = getConsumerGroup(groupId);
        group->addConsumer(consumer);
    }

    void deregisterConsumer(const std::string& groupId, Consumer* consumer) {
        auto* group = getConsumerGroup(groupId);
        group->removeConsumer(consumer);
    }

    bool hasTopic(const std::string& name) const {
        return topics_.count(name) > 0;
    }
};

"""
Demo: Mini Pub/Sub Messaging System (Mini-Kafka)
"""
import time
import threading


def main():
    # 1. Create the broker
    broker = PubSubBroker()

    # 2. Create a topic with 3 partitions
    orders_topic = broker.create_topic("orders", 3)
    print(f"Created: {orders_topic}")

    # 3. Create two consumer groups (independent consumption)
    analytics_group = broker.create_consumer_group("analytics-group")
    notification_group = broker.create_consumer_group("notification-group")

    # 4. Create consumers and register them
    analytics_consumer1 = Consumer("analytics-1", "analytics-group")
    analytics_consumer2 = Consumer("analytics-2", "analytics-group")
    notif_consumer = Consumer("notif-1", "notification-group")

    broker.register_consumer("analytics-group", analytics_consumer1)
    broker.register_consumer("analytics-group", analytics_consumer2)
    broker.register_consumer("notification-group", notif_consumer)

    # 5. Subscribe groups to the topic
    broker.subscribe("analytics-group", "orders")
    broker.subscribe("notification-group", "orders")

    # 6. Produce messages with keys
    producer = Producer("order-service")
    producer.publish(orders_topic, "user-123", "Order #1001 placed")
    producer.publish(orders_topic, "user-456", "Order #1002 placed")
    producer.publish(orders_topic, "user-123", "Order #1001 shipped")
    producer.publish(orders_topic, "user-789", "Order #1003 placed")
    producer.publish(orders_topic, None, "Broadcast: flash sale started")
    print("\nPublished 5 messages to 'orders' topic")

    # 7. Analytics group consumption
    print("\n--- Analytics Group ---")
    for p in analytics_group.get_assigned_partitions(analytics_consumer1):
        msgs = analytics_consumer1.poll(p, 10)
        print(f"  {analytics_consumer1.id} from {p}: {msgs}")

    for p in analytics_group.get_assigned_partitions(analytics_consumer2):
        msgs = analytics_consumer2.poll(p, 10)
        print(f"  {analytics_consumer2.id} from {p}: {msgs}")

    # 8. Notification group consumption
    print("\n--- Notification Group (independent progress) ---")
    for p in notification_group.get_assigned_partitions(notif_consumer):
        msgs = notif_consumer.poll(p, 10)
        print(f"  {notif_consumer.id} from {p}: {msgs}")
        notification_group.commit_offset(p, notif_consumer.get_offset(p))

    # 9. Re-poll (should be empty)
    print("\n--- Re-poll notification group (should be empty) ---")
    for p in notification_group.get_assigned_partitions(notif_consumer):
        msgs = notif_consumer.poll(p, 10)
        print(f"  {notif_consumer.id} from {p}: {msgs}")

    # 10. Replay via seek
    print("\n--- Replay: seek notification consumer to offset 0 ---")
    for p in notification_group.get_assigned_partitions(notif_consumer):
        notif_consumer.seek(p, 0)
        msgs = notif_consumer.poll(p, 10)
        print(f"  REPLAY {p}: {msgs}")


if __name__ == "__main__":
    main()

#include <iostream>
#include <string>
#include "PubSubBroker.hpp"
#include "Producer.hpp"

int main() {
    // 1. Create the broker
    PubSubBroker broker;

    // 2. Create a topic with 3 partitions
    Topic* ordersTopic = broker.createTopic("orders", 3);
    std::cout << "Created topic 'orders' with 3 partitions\n";

    // 3. Create two consumer groups
    ConsumerGroup* analyticsGroup = broker.createConsumerGroup("analytics-group");
    ConsumerGroup* notifGroup = broker.createConsumerGroup("notification-group");

    // 4. Create consumers and register them
    Consumer analyticsConsumer1("analytics-1", "analytics-group");
    Consumer analyticsConsumer2("analytics-2", "analytics-group");
    Consumer notifConsumer("notif-1", "notification-group");

    broker.registerConsumer("analytics-group", &analyticsConsumer1);
    broker.registerConsumer("analytics-group", &analyticsConsumer2);
    broker.registerConsumer("notification-group", &notifConsumer);

    // 5. Subscribe groups to the topic
    broker.subscribe("analytics-group", "orders");
    broker.subscribe("notification-group", "orders");

    // 6. Produce messages
    Producer producer("order-service");
    producer.publish(ordersTopic, "user-123", "Order #1001 placed");
    producer.publish(ordersTopic, "user-456", "Order #1002 placed");
    producer.publish(ordersTopic, "user-123", "Order #1001 shipped");
    producer.publish(ordersTopic, "user-789", "Order #1003 placed");
    producer.publish(ordersTopic, "", "Broadcast: flash sale started");
    std::cout << "\nPublished 5 messages to 'orders' topic\n";

    // 7. Analytics group - each consumer reads assigned partitions
    std::cout << "\n--- Analytics Group ---\n";
    for (auto* p : analyticsGroup->getAssignedPartitions(&analyticsConsumer1)) {
        auto msgs = analyticsConsumer1.poll(p, 10);
        std::cout << "  " << analyticsConsumer1.getId() << " from partition-"
                  << p->getId() << ": ";
        for (const auto& m : msgs) std::cout << m.toString() << " ";
        std::cout << "\n";
    }
    for (auto* p : analyticsGroup->getAssignedPartitions(&analyticsConsumer2)) {
        auto msgs = analyticsConsumer2.poll(p, 10);
        std::cout << "  " << analyticsConsumer2.getId() << " from partition-"
                  << p->getId() << ": ";
        for (const auto& m : msgs) std::cout << m.toString() << " ";
        std::cout << "\n";
    }

    // 8. Notification group
    std::cout << "\n--- Notification Group ---\n";
    for (auto* p : notifGroup->getAssignedPartitions(&notifConsumer)) {
        auto msgs = notifConsumer.poll(p, 10);
        std::cout << "  " << notifConsumer.getId() << " from partition-"
                  << p->getId() << ": ";
        for (const auto& m : msgs) std::cout << m.toString() << " ";
        std::cout << "\n";
        notifGroup->commitOffset(p, notifConsumer.getOffset(p));
    }

    // 9. Re-poll (empty)
    std::cout << "\n--- Re-poll (should be empty) ---\n";
    for (auto* p : notifGroup->getAssignedPartitions(&notifConsumer)) {
        auto msgs = notifConsumer.poll(p, 10);
        std::cout << "  " << notifConsumer.getId() << " from partition-"
                  << p->getId() << ": " << msgs.size() << " messages\n";
    }

    // 10. Replay
    std::cout << "\n--- Replay via seek ---\n";
    for (auto* p : notifGroup->getAssignedPartitions(&notifConsumer)) {
        notifConsumer.seek(p, 0);
        auto msgs = notifConsumer.poll(p, 10);
        std::cout << "  REPLAY partition-" << p->getId() << ": ";
        for (const auto& m : msgs) std::cout << m.toString() << " ";
        std::cout << "\n";
    }

    return 0;
}