aio-libs / aio-libs/aiokafka

Producer.send_and_wait() hangs up forever if `sender_task` is cancelled

Offen
#1,039 0 Kommentare 4 Reaktionen 0 zugewiesene Personen Auf GitHub ansehen
Vorherrschende Sprache
Python
Sterne
1.4k
Forks
269
Ø Merge
1 T. 1 Std.
Gemergte PRs (30 T.)
6

Beschreibung

aiokafka [relies on `ioloop.create_task()`](https://github.com/aio-libs/aiokafka/blob/v0.11.0/aiokafka/producer/sender.py#L76) method, which run coroutine in background and makes it "cancelable". The problem that [the tasks list in asyncio is public](https://docs.python.org/3/library/asyncio-task.html#asyncio.all_tasks) and external code can cancel `sender_task`. After that `Producer` will be in the state, when any message will be successfully added to `message_accumulator`, but will never actually sent by `Sender`.

Originally I faced with this problem when using `AsyncTestCase` from Tornado. After each test, Tornado [cancels all running background tasks](https://github.com/tornadoweb/tornado/blob/v6.4.1/tornado/testing.py#L170-L190). It's could be questionable how legal this behavior, but the fact is that only the public APIs are used by Tornado and there are no errors or exceptions are logged from aiokafka side. It just hangs up forever on the second and next tests.

I haven't digged in other circumstances when background asyncio tasks could be canceled by third-party libs, but if this happens there should be some sort of exception or error logging.

**Expected behaviour**
I expect the `Sender` recover after stop either `Producer` raise an error what `sender_task` is canceled and message will never delivered.

**Environment (please complete the following information):**
- aiokafka version: 0.11.0
- Kafka Broker version: 3.7.0

**Reproducible example**
```python
import aiokafka
import asyncio

def cancel_tasks(loop):
tasks = [t for t in asyncio.all_tasks(loop) if not t.done()]
for t in tasks:
if not t.done():
t.cancel()
if tasks:
done, pending = loop.run_until_complete(asyncio.wait(tasks))
assert not pending
return tasks

async def prepare():
global producer
producer = aiokafka.AIOKafkaProducer(bootstrap_servers='kafka')
await producer.start()

async def send():
print('>>> before send')
resp = await producer.send_and_wait('topic', b'body')
print('>>>', resp)
await producer.stop()

loop = asyncio.get_event_loop()
loop.run_until_complete(prepare())
print('>>>', cancel_tasks(loop))
loop.run_until_complete(send())
```

Beitragsleitfaden

Beitragsleitfaden öffnen

Bewertung

Dieses Issue wurde noch nicht bewertet.

Neue Issues direkt in Ihr Postfach

Eine kurze Übersicht über anfängerfreundliche GitHub-Issues.