opentelemetry_aiokafka_instrumentation_producer_consumer_tracing_quickstart.py

1import asyncio
2from aiokafka import AIOKafkaProducer, AIOKafkaConsumer
3from opentelemetry.instrumentation.aiokafka import AIOKafkaInstrumentor
4
5# The instrument() method enables automatic tracing for aiokafka
6AIOKafkaInstrumentor().instrument()
7
8async def produce():
9    producer = AIOKafkaProducer(bootstrap_servers='localhost:9092')
10    await producer.start()
11    try:
12        # A span will be automatically created for this send operation
13        await producer.send_and_wait("my-topic", b"Hello, OpenTelemetry!")
14    finally:
15        await producer.stop()
16
17async def consume():
18    consumer = AIOKafkaConsumer(
19        "my-topic",
20        bootstrap_servers='localhost:9092',
21        group_id="my-group"
22    )
23    await consumer.start()
24    try:
25        # Spans will be automatically created when messages are received
26        async for msg in consumer:
27            print(f"Consumed: {msg.value}")
28            break 
29    finally:
30        await consumer.stop()
31
32async def main():
33    await produce()
34    await consume()
35
36if __name__ == "__main__":
37    asyncio.run(main())