Mock Kafka Consumer Group Partition Assignment in Python

from collections import defaultdict


class ConsumerGroupAssignment:
    def __init__(self, group_name, topics_partitions):
        self.group_name = group_name
        self.consumers = {}
        self.assignments = defaultdict(set)
        topics_partitions = sorted(
            [(topic, partition) for topic, partitions in topics_partitions.items() for partition in partitions],
            key=lambda x: (x[0], x[1])
        )

        consumer_index = 0
        consumer_count = 0
        for topic, partition in topics_partitions:
            consumer = f"{group_name}-consumer-{consumer_index}"
            self.consumers[consumer] = consumer_index
            self.assignments[consumer].add((topic, partition))
            consumer_count = len(self.consumers)
            consumer_index = (consumer_index + 1) % (consumer_count if consumer_count > 0 else 1)

    def get_assignment(self, consumer):
        return sorted(self.assignments.get(consumer, set()))

    def get_all_assignments(self):
        result = {}
        for consumer in sorted(self.consumers):
            result[consumer] = self.get_assignment(consumer)
        return result


if __name__ == "__main__":
    topics = {
        "orders": [0, 1, 2],
        "payments": [0, 1],
        "notifications": [0]
    }
    mock_group = ConsumerGroupAssignment("order-service-1", topics)
    for consumer, assignment in mock_group.get_all_assignments().items():
        print(f"{consumer}: {assignment}")