aio-pika queue iterator drops context on start of iteration

Describe your environment

OS: MacOS, but works the same on Ubuntu
Python version: Python 3.11
Package version: (e.g., 0.48.0)

What happened?

I'm get message from queue using async queue.iterator

        async with queue.iterator() as queue_iter:
            async for message in queue_iter:
                # http call here will have no parent span

But inside context manager there is no context and if I get current span it has trace_id=0x00000000000000000000000000000000

Steps to Reproduce

Dependancies

aio-pika = "^9.0.7"
opentelemetry-api = "^1.27.0"
opentelemetry-sdk = "^1.27.0"
opentelemetry-instrumentation = "^0.48b0"
opentelemetry-instrumentation-aio-pika = "^0.48b0"

Run script

import asyncio

from aio_pika import Message, connect_robust
from opentelemetry import trace
from opentelemetry.instrumentation.aio_pika import AioPikaInstrumentor
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
from opentelemetry.trace import get_current_span

provider = TracerProvider()
processor = BatchSpanProcessor(InMemorySpanExporter())
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)
AioPikaInstrumentor().instrument()


async def main():
    # Connection
    connection = await connect_robust("amqp://guest:guest@localhost/")

    async with connection:
        channel = await connection.channel()
        queue = await channel.declare_queue("test_queue", durable=True)
        message_body = b"Hello, RabbitMQ!"
        await channel.default_exchange.publish(
            Message(message_body),
            routing_key=queue.name,
        )

        print(f"Sent message: {message_body.decode()}")

        async with queue.iterator() as queue_iter:
            async for message in queue_iter:
                print(get_current_span())
                # http call here will have no parent span
                await message.ack()


if __name__ == "__main__":
    asyncio.run(main())

Also I added debug prints in callback decorator here

    def decorate(
        self, callback: Callable[[AbstractIncomingMessage], Any]
    ) -> Callable[[AbstractIncomingMessage], Any]:
        async def decorated(message: AbstractIncomingMessage):
            if not is_instrumentation_enabled():
                return await callback(message)
            headers = message.headers or {}
            ctx = propagate.extract(headers)
            token = context.attach(ctx)
            span = self._get_span(message)
            if not span:
                return await callback(message)
            try:
                with trace.use_span(span, end_on_exit=True) as span:
                    print("use span", span)
                    print("Callback is", callback)
                    return_value = await callback(message)
            finally:
                print("Finally detached")
                context.detach(token)
            return return_value

        return decorated

Expected Result

Span inside async for message in queue_iter propagate parent id correctly

Actual Result

Print outputs

Sent message: Hello, RabbitMQ!
Span inside queue iter NonRecordingSpan(SpanContext(trace_id=0x00000000000000000000000000000000, span_id=0x0000000000000000, trace_flags=0x00, trace_state=[], is_remote=False))
Span inside decorator with trace.use_span _Span(name="test_queue receive", context=SpanContext(trace_id=0x38c7e644aa93e08293bbc1525e6ddc70, span_id=0x7ad80d8eb7aa7725, trace_flags=0x01, trace_state=[], is_remote=False))
Callback is <bound method QueueIterator.on_message of <RobustQueueIterator: queue='test_queue' ctag='ctag1.d18fd06611614d60a6fa4d3bb6cab661'>>
Finally detached
Span inside message NonRecordingSpan(SpanContext(trace_id=0x00000000000000000000000000000000, span_id=0x0000000000000000, trace_flags=0x00, trace_state=[], is_remote=False))

Span inside async for message in queue_iter has no context, because it was dropped in finally block

Additional context

For the context: QueueIterator.on_message is a callback which is decorated by instrumentator

My assumption what is happening:

• async with queue.iterator() as queue_iter calls QueueIterator.__aenter__
• __aenter__ calls QueueIterator.consume, and consume eventually calls QueueIterator.on_message
• on_message put message in internal async Queue and return
• On this point finally block of instrumentator callback drop the context
• async for message in queue_iter calls QueueIterator.__anext__
• __anext__ get message from internal queue and return it. Context already dropped on this stage

Since context is dropped, span inside async for message in queue_iter has trace_id=0x00000000000000000000000000000000

Would you like to implement a fix?

None

Start with instrumentation/opentelemetry-instrumentation-aio-pika/src/opentelemetry/instrumentation/aio_pika/callback_decorator.py and trace QueueIterator.on_message through QueueIterator.aenter, consume, and anext. Reproduce the example with the supplied dependencies, then add or update tests to verify that context remains available during async iteration and is detached at the appropriate boundary.