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

[homm]

aiokafka relies on ioloop.create_task() method, which run coroutine in background and makes it "cancelable". The problem that the tasks list in asyncio is public 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. 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

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