aio-libs / aio-libs/aiokafka

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

Đang mở
#1,039 0 bình luận 4 reaction 0 người được giao Xem trên GitHub
Ngôn ngữ chính
Python
Star
1.4k
Fork
269
Merge trung bình
1 ngày 1 giờ
Pull request đã merge (30 ngày)
6

Mô tả

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

Hướng dẫn đóng góp

Mở hướng dẫn đóng góp

Đánh giá

Issue này chưa được đánh giá.

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.