crash after upgrading Faust to v0.10.0 + Mode to v0.3.1

[joonjeong]

Checklist

• [x] I have included information about relevant versions
• [ ] I have verified that the issue persists when using the master branch of Faust.

Steps to reproduce

After I upgrade faust-streaming==0.10.0, it goes crash like this.

Expected behavior

Tell us what you expected to happen.

Actual behavior

Tell us what happened instead.

Full traceback

[2022-11-30 14:44:27,908] [6] [INFO] [^Worker]: Starting... 
[2022-11-30 14:44:27,912] [6] [INFO] [^-App]: Starting... 
[2022-11-30 14:44:27,913] [6] [INFO] [^--PrometheusMonitor]: Starting... 
[2022-11-30 14:44:27,913] [6] [INFO] [^--Producer]: Starting... 
[2022-11-30 14:44:27,913] [6] [INFO] [^---ProducerBuffer]: Starting... 
[2022-11-30 14:44:27,918] [6] [WARNING] Sending username and password in the clear 
[2022-11-30 14:44:27,919] [6] [INFO] Authenticated as user via PLAIN 
[2022-11-30 14:44:27,923] [6] [WARNING] Sending username and password in the clear 
[2022-11-30 14:44:27,924] [6] [INFO] Authenticated as user via PLAIN 
[2022-11-30 14:44:27,928] [6] [INFO] [^--ThreadedProducer]: Starting... 
[2022-11-30 14:44:27,930] [6] [ERROR] [^--ThreadedProducer]: Crashed reason=RuntimeError("Task <Task pending name='Task-12' coro=<Event.wait() running at /usr/lib/python3.8/asyncio/locks.py:309> cb=[_release_waiter(<Future pendi...fe590bbe0>()]>)() at /usr/lib/python3.8/asyncio/tasks.py:429]> got Future <Future pending> attached to a different loop") 
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/dist-packages/mode/services.py", line 843, in _execute_task
    await task
  File "/usr/local/lib/python3.8/dist-packages/mode/threads.py", line 182, in _keepalive2
    await self.sleep(1.1)
  File "/usr/local/lib/python3.8/dist-packages/mode/services.py", line 696, in sleep
    await asyncio.wait_for(
  File "/usr/lib/python3.8/asyncio/tasks.py", line 494, in wait_for
    return fut.result()
  File "/usr/lib/python3.8/asyncio/locks.py", line 309, in wait
    await fut
RuntimeError: Task <Task pending name='Task-12' coro=<Event.wait() running at /usr/lib/python3.8/asyncio/locks.py:309> cb=[_release_waiter(<Future pendi...fe590bbe0>()]>)() at /usr/lib/python3.8/asyncio/tasks.py:429]> got Future <Future pending> attached to a different loop
[2022-11-30 14:44:27,931] [6] [ERROR] [^Worker]: Error: RuntimeError("Task <Task pending name='Task-12' coro=<Event.wait() running at /usr/lib/python3.8/asyncio/locks.py:309> cb=[_release_waiter(<Future pendi...fe590bbe0>()]>)() at /usr/lib/python3.8/asyncio/tasks.py:429]> got Future <Future pending> attached to a different loop") 
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/dist-packages/mode/worker.py", line 279, in execute_from_commandline
    self.loop.run_until_complete(self._starting_fut)
  File "/usr/lib/python3.8/asyncio/base_events.py", line 616, in run_until_complete
    return future.result()
  File "/usr/local/lib/python3.8/dist-packages/mode/services.py", line 800, in start
    await self._default_start()
  File "/usr/local/lib/python3.8/dist-packages/mode/services.py", line 807, in _default_start
    await self._actually_start()
  File "/usr/local/lib/python3.8/dist-packages/mode/services.py", line 831, in _actually_start
    await child.maybe_start()
  File "/usr/local/lib/python3.8/dist-packages/mode/services.py", line 859, in maybe_start
    await self.start()
  File "/usr/local/lib/python3.8/dist-packages/mode/services.py", line 800, in start
    await self._default_start()
  File "/usr/local/lib/python3.8/dist-packages/mode/services.py", line 807, in _default_start
    await self._actually_start()
  File "/usr/local/lib/python3.8/dist-packages/mode/services.py", line 831, in _actually_start
    await child.maybe_start()
  File "/usr/local/lib/python3.8/dist-packages/mode/threads.py", line 155, in maybe_start
    await self.start()
  File "/usr/local/lib/python3.8/dist-packages/mode/threads.py", line 176, in start
    await self._thread_running
  File "/usr/local/lib/python3.8/dist-packages/mode/services.py", line 843, in _execute_task
    await task
  File "/usr/local/lib/python3.8/dist-packages/mode/threads.py", line 182, in _keepalive2
    await self.sleep(1.1)
  File "/usr/local/lib/python3.8/dist-packages/mode/services.py", line 696, in sleep
    await asyncio.wait_for(
  File "/usr/lib/python3.8/asyncio/tasks.py", line 494, in wait_for
    return fut.result()
  File "/usr/lib/python3.8/asyncio/locks.py", line 309, in wait
    await fut
RuntimeError: Task <Task pending name='Task-12' coro=<Event.wait() running at /usr/lib/python3.8/asyncio/locks.py:309> cb=[_release_waiter(<Future pendi...fe590bbe0>()]>)() at /usr/lib/python3.8/asyncio/tasks.py:429]> got Future <Future pending> attached to a different loop

Versions

• Python version: 3.8.10
• Faust version: faust-streaming==0.10.0
• Operating system: ubuntu 20.04
• Kafka version
• RocksDB version (if applicable)

[wbarnha]

This is definitely a mode-streaming issue, sorry to see you're having this issue. I'll look into it. In the meantime I recommend using Faust v0.9.5.

[wbarnha]

@joonjeong could you help me reproduce this issue? I hope this fix will address the issue:

pip install --upgrade mode-streaming==0.3.2rc0

[wbarnha]

I'll keep this open until the issue is reported to be gone @joonjeong.

[wbarnha]

The bug still exists sadly. Encountered it again while testing Faust.

[lqhuang]

So how #35 fix this bug?

[joonjeong]

@wbarnha I checked mode-streaming==0.3.2 and faust-streaming==0.10.0 works fine!

[wbarnha]

The reason mode-streaming was failing was because we replaced .utils.locks import Event with from asyncio.locks import Event. Once you change those back and forth and run any Faust example, such as faust -A examples.windowed_aggregation -l info worker, you'll see the error come back immediately.