aiokafka: cannot trace message consuming with auto-instrumentation

[andreaturli]

What problem do you want to solve?

When running https://aiokafka.readthedocs.io/en/stable/#aiokafkaconsumer, traces are not produced when using zero-code instrumentation similar to https://opentelemetry.io/docs/languages/python/getting-started/#instrumentation

Describe the solution you'd like

I'd like aiokafka consumer to produce traces with zero-code instrumentation

Describe alternatives you've considered

Having something like the following

async def consume():
    consumer = AIOKafkaConsumer(
        'my_topic', 'my_other_topic',
        bootstrap_servers='localhost:9092',
        group_id="my-group")
    # Get cluster layout and join group `my-group`
    await consumer.start()
    try:
        # Consume messages
        async for msg in consumer:
            print("consumed: ", msg.topic, msg.partition, msg.offset,
                  msg.key, msg.value, msg.timestamp)
    finally:
        # Will leave consumer group; perform autocommit if enabled.
        await consumer.stop()

I've tried to get traces using

async def main():
   consume()

and also

async def main():
   with tracer.start_as_current_span("kafka-consumer") as span:
      consume()

but nothing gets created in my OTEL exporter

Additional Context

Using

opentelemetry-instrumentation-aiokafka==0.53b1 and python 3.13

Would you like to implement a fix?

None

[emdneto]

I just tested this with the examples from #3466, and it seems to be working fine.

[andreaturli]

I can confirm it seems to be working when running something like opentelemetry-instrument python consumer.py where consumer.py is

import asyncio
import json
import logging

from aiokafka import AIOKafkaProducer, AIOKafkaConsumer

# Configure logging
logging.basicConfig(
    level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

async def send_one_message(topic_name, message, bootstrap_servers='localhost:9092'):
    """
    Send a single message to a Kafka topic

    Args:
        topic_name (str): Name of the Kafka topic
        message (str or bytes): Message to send
        bootstrap_servers (str): Kafka broker(s) address
    """
    # Convert string message to bytes if needed
    if isinstance(message, str):
        message = message.encode('utf-8')

    # Create producer
    producer = AIOKafkaProducer(bootstrap_servers=bootstrap_servers)

    try:
        # Start the producer
        await producer.start()

        # Send message
        logger.info(f"Sending message to topic {topic_name}")
        await producer.send(topic_name, message)
        logger.info(f"Message sent successfully")

    finally:
        # Stop the producer
        await producer.stop()

async def consume_messages():
    # Create a consumer instance
    consumer = AIOKafkaConsumer(
        'my-topic',  # topic name
        bootstrap_servers='localhost:9092',  # Kafka broker(s)
        group_id='my-group',  # Consumer group ID
        value_deserializer=lambda m: json.loads(m.decode('utf-8'))  # Deserialize JSON messages
    )

    # Start the consumer
    await consumer.start()

    try:
        # Consume messages
        async for msg in consumer:
            print(f"Received message:")
            print(f"Topic: {msg.topic}")
            print(f"Partition: {msg.partition}")
            print(f"Offset: {msg.offset}")
            print(f"Key: {msg.key}")
            print(f"Value: {msg.value}")
            print("------------------------")


    except Exception as e:
        logger.error(f"Error: {e}")
    finally:
        # Stop the consumer
        await consumer.stop()

# Example usage
if __name__ == "__main__":
    # Topic name
    TOPIC = "my-topic"
    # Message to send
    MESSAGE = "Hello, Kafka!"
    # Send a single message
    asyncio.run(send_one_message(TOPIC, MESSAGE))
    # Consume messages
    asyncio.run(consume_messages())

with

> pip freeze | grep opentelemetry
opentelemetry-api==1.32.1
opentelemetry-distro==0.53b1
opentelemetry-exporter-otlp==1.32.1
opentelemetry-exporter-otlp-proto-common==1.32.1
opentelemetry-exporter-otlp-proto-grpc==1.32.1
opentelemetry-exporter-otlp-proto-http==1.32.1
opentelemetry-instrumentation==0.53b1
opentelemetry-instrumentation-aiohttp-client==0.53b1
opentelemetry-instrumentation-aiohttp-server==0.53b1
opentelemetry-instrumentation-aiokafka==0.53b1
opentelemetry-instrumentation-asyncio==0.53b1
opentelemetry-instrumentation-asyncpg==0.53b1
opentelemetry-instrumentation-dbapi==0.53b1
opentelemetry-instrumentation-grpc==0.53b1
opentelemetry-instrumentation-logging==0.53b1
opentelemetry-instrumentation-requests==0.53b1
opentelemetry-instrumentation-sqlite3==0.53b1
opentelemetry-instrumentation-threading==0.53b1
opentelemetry-instrumentation-urllib==0.53b1
opentelemetry-instrumentation-urllib3==0.53b1
opentelemetry-instrumentation-wsgi==0.53b1
opentelemetry-proto==1.32.1
opentelemetry-sdk==1.32.1
opentelemetry-semantic-conventions==0.53b1
opentelemetry-util-http==0.53b1

see also the traces on my exporter

---

not sure why when used in combination with other libraries, namely aiohttp-server and asyncpg, it gives me this weird message

Any idea?

[artkravchenko]

Looks like the consumer span only exists in the global instrumentator's async_consume_hook.

When we use AIOKafkaConsumer::__anext__, the consumer span is no longer there,
so any downstream operations don't have any context.

from opentelemetry import trace

...

async for msg in consumer:
    # seeing the traceparent header in the consumed message from the producer
    msg.headers  # `[..., ('traceparent', b'00-24770f968d52edaf2771745749cec452-9c407f42b9cdf30f-01')]`

    # the consumer span is created and sent to the span processor, visible in `ConsoleSpanExporter()`

    # but the consumer span gets ended too early, so it's missing in the actual consumer logic here
    span = trace.get_current_span()  # NonRecordingSpan(SpanContext(trace_id=0x00000000000000000000000000000000, span_id=0x0000000000000000, trace_flags=0x00, trace_state=[], is_remote=False))

    ... # so db queries, http calls aren't attached to the consumer span