How to Mock a Kafka Producer Batch Send in Python

import json
import random
import time
from datetime import datetime


class MockKafkaProducer:
    def __init__(self, topic):
        self.topic = topic
        self.sent_messages = []

    def send(self, value, key=None):
        message = {
            "topic": self.topic,
            "key": key,
            "value": value,
            "timestamp": datetime.utcnow().isoformat(),
            "partition": random.randint(0, 3)
        }
        self.sent_messages.append(message)
        time.sleep(0.01)  # Simulate network latency
        return len(self.sent_messages) - 1

    def flush(self):
        print(f"Flushing {len(self.sent_messages)} messages to topic '{self.topic}'")
        for msg in self.sent_messages:
            print(f"  [partition {msg['partition']}] {msg['timestamp']}: {msg['value']}")


def main():
    producer = MockKafkaProducer("user-events")

    batch_size = 10
    events = [
        {"user_id": i, "action": random.choice(["click", "view", "purchase"]), "amount": random.randint(1, 100)}
        for i in range(1, batch_size + 1)
    ]

    batch_ids = []
    for event in events:
        batch_ids.append(producer.send(json.dumps(event), key=str(event["user_id"])))

    producer.flush()

    print(f"\nBatch of {len(batch_ids)} messages sent successfully.")
    print(f"Sample event: {events[0]}")


if __name__ == "__main__":
    main()