How to Mock a Kafka Rebalance Listener in Python

import time
from collections import defaultdict


class MockKafkaConsumer:
    def __init__(self):
        self.assignments = defaultdict(list)
        self.rebalances = 0

    def assign(self, partitions):
        self.rebalances += 1
        self.assignments.clear()
        for partition in partitions:
            self.assignments[partition.topic].append(partition.partition)
        print(f"Rebalance #{self.rebalances}: assigned {dict(self.assignments)}")

    def poll(self):
        time.sleep(0.1)


class RebalanceListener:
    def on_partitions_revoked(self, consumer):
        print("Partitions revoked, pausing processing...")
        consumer.assignments.clear()

    def on_partitions_assigned(self, consumer, partitions):
        print("New partitions assigned, starting processing...")
        consumer.assign(partitions)


class Partition:
    def __init__(self, topic, partition):
        self.topic = topic
        self.partition = partition


def main():
    consumer = MockKafkaConsumer()
    listener = RebalanceListener()

    initial_partitions = [Partition("orders", 0), Partition("orders", 1)]
    listener.on_partitions_revoked(consumer)
    listener.on_partitions_assigned(consumer, initial_partitions)

    new_partitions = [Partition("orders", 1), Partition("orders", 2)]
    listener.on_partitions_revoked(consumer)
    listener.on_partitions_assigned(consumer, new_partitions)


if __name__ == "__main__":
    main()