How to Create a Mock Kafka Producer in Python

import json
import time
from kafka import KafkaProducer
from kafka.errors import KafkaError

def create_mock_producer(bootstrap_servers="localhost:9092", topic="input-topic"):
    """Create a Kafka producer that generates mock streaming data."""
    producer = KafkaProducer(
        bootstrap_servers=bootstrap_servers,
        value_serializer=lambda v: json.dumps(v).encode("utf-8")
    )

    def send_mock_record(record_id):
        mock_record = {
            "id": record_id,
            "timestamp": time.time(),
            "value": record_id * 10,
            "status": "active" if record_id % 2 == 0 else "inactive"
        }
        try:
            future = producer.send(topic, value=mock_record)
            future.get(timeout=10)
            print(f"Sent: {json.dumps(mock_record)}")
        except KafkaError as e:
            print(f"Failed to send record {record_id}: {e}")

    return send_mock_record

if __name__ == "__main__":
    send_mock = create_mock_producer()

    # Simulate streaming a few records
    for i in range(1, 6):
        send_mock(i)
        time.sleep(0.5)

    print("Mock streaming completed.")