confluent_kafka_producer_with_type_hints_and_callback.py

1from confluent_kafka import Producer
2import socket
3
4# Configuration dictionary for the producer
5# The stubs allow type checkers to validate these keys/values
6conf = {
7    'bootstrap.servers': 'localhost:9092',
8    'client.id': socket.gethostname()
9}
10
11# Initialize Producer
12# Typing: Producer class is defined in the stubs
13producer = Producer(conf)
14
15def acked(err, msg):
16    """ Delivery report callback called once for each message. """
17    if err is not None:
18        print(f"Failed to deliver message: {err}")
19    else:
20        print(f"Message delivered to {msg.topic()} [{msg.partition()}]")
21
22# Produce a message
23# Typing: .produce() arguments are type-checked via the stubs
24producer.produce(
25    topic='test_topic', 
26    key='key', 
27    value='value', 
28    callback=acked
29)
30
31# Wait for any outstanding messages to be delivered
32producer.flush()