Confluent Kafka Python Client

import os
from confluent_kafka import Producer
from confluent_kafka import Consumer, KafkaException

def delivery_report(err, msg):
    if err is not None:
        print(f'Message delivery failed: {err}')
    else:
        print(f'Message delivered to {msg.topic()} [{msg.partition()}]')

def run_producer():
    conf = {
        'bootstrap.servers': os.environ.get('KAFKA_BOOTSTRAP_SERVERS', 'localhost:9092'),
        # Add security configs if connecting to Confluent Cloud or secured clusters
        # 'security.protocol': 'SASL_SSL',
        # 'sasl.mechanisms': 'PLAIN',
        # 'sasl.username': os.environ.get('KAFKA_API_KEY'),
        # 'sasl.password': os.environ.get('KAFKA_API_SECRET')
    }

    producer = Producer(conf)
    topic = 'my_test_topic'

    try:
        for i in range(5):
            message = f'Hello Kafka from Python {i}'
            producer.produce(topic, message.encode('utf-8'), callback=delivery_report)
            print(f'Produced message: {message}')
        producer.flush(10) # Wait up to 10 seconds for messages to be delivered

    except BufferError:
        print('Local producer queue is full. Try again later.')
    except Exception as e:
        print(f'Error producing message: {e}')
    finally:
        producer.flush()

def run_consumer():
    conf = {
        'bootstrap.servers': os.environ.get('KAFKA_BOOTSTRAP_SERVERS', 'localhost:9092'),
        'group.id': 'my_python_consumer_group',
        'auto.offset.reset': 'earliest',
        'enable.auto.commit': False,
        # Add security configs if connecting to Confluent Cloud or secured clusters
        # 'security.protocol': 'SASL_SSL',
        # 'sasl.mechanisms': 'PLAIN',
        # 'sasl.username': os.environ.get('KAFKA_API_KEY'),
        # 'sasl.password': os.environ.get('KAFKA_API_SECRET')
    }

    consumer = Consumer(conf)
    topic = 'my_test_topic'

    try:
        consumer.subscribe([topic])

        while True:
            msg = consumer.poll(1.0) # Poll for messages with a 1-second timeout
            if msg is None:
                continue
            if msg.error():
                if msg.error().code() == KafkaException.PARTITION_EOF:
                    # End of partition event, not an error
                    print(f'{msg.topic()} [{msg.partition()}] reached end offset {msg.offset()}')
                else:
                    print(f'Consumer error: {msg.error()}')
            else:
                print(f"Received message: {msg.value().decode('utf-8')} from {msg.topic()} [{msg.partition()}] @ offset {msg.offset()}")
                consumer.commit(msg)

    except KeyboardInterrupt:
        pass
    except Exception as e:
        print(f'Error consuming message: {e}')
    finally:
        consumer.close()

if __name__ == '__main__':
    print("Running producer...")
    run_producer()
    print("\nRunning consumer (Press Ctrl+C to stop)...")
    run_consumer()