RuntimeError: There is no current event loop in thread 'Thread-X'

[kgaikwad]

Describe the bug
I am trying to update aiokafa version from 0.6.0 to 0.7.0. I am getting an exception - RuntimeError: There is no current event loop in thread 'Thread-X'

Expected behaviour
I expect it should run without an error.

Environment (please complete the following information):

• aiokafka version (python -c "import aiokafka; print(aiokafka.__version__)"): 0.7.0
• kafka-python version (python -c "import kafka; print(kafka.__version__)"): 2.0.2
• Kafka Broker version (kafka-topics.sh --version):
• Other information (Confluent Cloud version, etc.):

I am using python version 3.6.11. On aiokafka version upgrade to 0.7.0, faced an issue with dataclasses module (Exception - ModuleNotFoundError: No module named ‘dataclasses’). To resolve this, I have explicitly added dataclasses = "==0.5" in my Pipfile. Issue got resolved but led to another issue related to event loop.

Error logs:

lib/python3.6/site-packages/aiokafka/util.py", line 68, in get_running_loop
    loop = asyncio.get_event_loop()
  File "/usr/lib64/python3.6/asyncio/events.py", line 694, in get_event_loop
    return get_event_loop_policy().get_event_loop()
  File "/usr/lib64/python3.6/asyncio/events.py", line 602, in get_event_loop
    % threading.current_thread().name)
RuntimeError: There is no current event loop in thread 'Thread-1'.

Any pointers?

[ods]

Could you provide complete traceback? The chunk here doesn't show the caller of get_running_loop, while it's supposed to be called from coroutine functions only.

[kgaikwad]

Thank you @ods for a quick response.

Here is traceback:

Exception in thread Thread-2:
Traceback (most recent call last):
  File "/usr/lib64/python3.6/threading.py", line 916, in _bootstrap_inner
    self.run()
  File "/usr/lib64/python3.6/threading.py", line 864, in run
    self._target(*self._args, **self._kwargs)
  File "{app_path}/report_processor.py", line 756, in asyncio_report_processor_thread
    processor = ReportProcessor()
  File "{app_path}/report_processor.py", line 115, in __init__
    loop=report_processing_loop, bootstrap_servers=KAFKA_ADDRESS
  File "{virtualenv_path}/lib/python3.6/site-packages/aiokafka/producer/producer.py", line 255, in __init__
    sasl_oauth_token_provider=sasl_oauth_token_provider)
  File "{virtualenv_path}/lib/python3.6/site-packages/aiokafka/client.py", line 151, in __init__
    self._md_update_waiter = create_future()
  File "{virtualenv_path}/lib/python3.6/site-packages/aiokafka/util.py", line 29, in create_future
    loop = get_running_loop()
  File "{virtualenv_path}/lib/python3.6/site-packages/aiokafka/util.py", line 68, in get_running_loop
    loop = asyncio.get_event_loop()
  File "/usr/lib64/python3.6/asyncio/events.py", line 694, in get_event_loop
    return get_event_loop_policy().get_event_loop()
  File "/usr/lib64/python3.6/asyncio/events.py", line 602, in get_event_loop
    % threading.current_thread().name)
RuntimeError: There is no current event loop in thread 'Thread-2'.

[tvoinarovskyi]

Hi @kgaikwad, sorry about the long wait. In the latest release, we introduced several implicit loop fixes, so that will get the loop that is currently running your code using "asyncio.get_running_loop()". Explicit loop argument behaviour is deprecated for most of asyncio high-level operations, so we had to clean that up, but maybe we overdid it a bit.
We will try to get a better implementation for the next release and fix the documentation to reflect the change. In the meantime, you can fix this by running your code inside an async function. So, instead of something like this:

producer = AIOKafkaProducer()

async def produce():
    await producer.start()
    await producer.produce()

asyncio.run(produce())

You would need to put the initialization inside of an async function too, so either:

async def produce():
    producer = AIOKafkaProducer()
    await producer.start()
    await producer.produce()

asyncio.run(produce())

Or using a separate factory:

async def init():
    return  AIOKafkaProducer()

def produce(producer):
    producer = AIOKafkaProducer()
    await producer.start()
    await producer.produce()

producer = asyncio.run()
asyncio.run(produce(producer))

Hope that helps 😓, please let us know if you have any more problems ^^