Using aiokafka with opentelemetry-instrumentation cause memory leak

[yonathan-wolloch-lendbuzz]

Describe the bug
As also mentioned in #3383, we have a memory leak (consistent linear increase in memory utilisation) when we use AIOKafkaConsumer in our fastapi code, it seems that the source is endless increase in objects from the path /python3.13/site-packages/opentelemetry/instrumentation/asyncio/init.py.

trying to use getone() within a while loop instead of anext reduce the leak pace but did not solve it. some of our checks showed that setting enable_auto_commit to False fixes this issue but we do want to use auto committing.

Expected behaviour
stable memory utilization when using aiokafka's best practices.

Environment (please complete the following information):

• aiokafka version (python -c "import aiokafka; print(aiokafka.__version__)"): 0.12.0
• Kafka Broker version (kafka-topics.sh --version): 3.6-IV2

Reproducible example

from aiokafka import AIOKafkaConsumer
import asyncio

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()

asyncio.run(consume())

[yonathan-wolloch-lendbuzz]

Setting the env var OTEL_PYTHON_DISABLED_INSTRUMENTATIONS=asyncio solves that issue, but we don't want to use this solution.

[vmaurin]

I have observed the same, but not sure it is fully related to aiokafka, more the python OTEL implementation is leaking when used in an asyncio context

[yonathan-wolloch-lendbuzz]

@vmaurin Is there a workaround we can use or different way for incorporating the consumer within the fastapi app to try and avoid the leakage? or the only way is using OTEL_PYTHON_DISABLED_INSTRUMENTATIONS=asyncio as we did?

[vmaurin]

Unfortunately we ended up removing the instrumentation completely as you suggested. I just wanted to point out that I don't think that is an issue specific to aiokafka but more otel python so probably open-telemetry/opentelemetry-python-contrib#3383 is the best place to have these discussion and maybe manage to have an example to reproduce the issue without aiokafka (like just "pure" asyncio" example)

[yonathan-wolloch-lendbuzz]

@vmaurin
👋 can you help opentelemetry-python-contrib reproduce the issue, or share light about why it is not reproducing in the test they conducted?

[yonathan-wolloch-lendbuzz]

@vmaurin @ods @dimastbk
👋

Final Issue Description

After extensive investigation, we've identified a memory leak involving the async def __coordination_routine(self) method in aiokafka.consumer.group_coordinator. This routine creates six futures, of which the following three:

• self._closing
• subscription.unsubscribe_future
• self._rejoin_needed_fut

continuously accumulate callbacks due to the interaction with AsyncioInstrumentor.trace_future from opentelemetry.instrumentation.asyncio. Each invocation of __coordination_routine adds three new callbacks to these futures indefinitely, causing unbounded memory growth.

Debugging Snippet

Here's a sanitized excerpt from our debug session highlighting the accumulated callbacks:

[
    <Future pending cb=[AsyncioInstrumentor.trace_future.<locals>.callback() at /path/to/project/.venv/lib/python3.13/site-packages/opentelemetry/instrumentation/asyncio/__init__.py:306, <70066 more callbacks>]>,
    <Future finished result=None>,
    <Future pending cb=[AsyncioInstrumentor.trace_future.<locals>.callback() at /path/to/project/.venv/lib/python3.13/site-packages/opentelemetry/instrumentation/asyncio/__init__.py:306, <70066 more callbacks>]>,
    <Future pending cb=[AsyncioInstrumentor.trace_future.<locals>.callback() at /path/to/project/.venv/lib/python3.13/site-packages/opentelemetry/instrumentation/asyncio/__init__.py:306, <70066 more callbacks>]>,
    <Task pending coro=<GroupCoordinator._heartbeat_routine() running at /path/to/project/.venv/lib/python3.13/site-packages/aiokafka/consumer/group_coordinator.py:773> wait_for=<Future pending cb=[Task.task_wakeup()]>>,
    <Task pending coro=<GroupCoordinator._commit_refresh_routine() running at /path/to/project/.venv/lib/python3.13/site-packages/aiokafka/consumer/group_coordinator.py:910> wait_for=<Future pending cb=[Task.task_wakeup()]>>
]

We'd greatly appreciate assistance or suggestions on resolving this issue.

[vmaurin]

Amazing job tracking down the issue !

I guess there could be two things we could do here :

• doing a minimal example without aiokafka so maybe opentelemetry can work on a mitigation (so creating future, waiting for one to be completed, not looking into the others)
• maybe in aiokafka, in this co-routine, trying to do something with the future still pending after the asyncio.wait call (cancelling them ?), but nothing in the usage of wait method says we should do some processing on these https://docs.python.org/3/library/asyncio-task.html#asyncio.wait

[yonathan-wolloch-lendbuzz]

I think its best waiting for opentelemetry-instrumentation-asyncio to release the fix and add a comment about using that version in the docs :)

[yonathan-wolloch-lendbuzz]

Fixed in opentelemetry-python-contrib Version 1.33.0/0.54b0