Consumer stopped consuming, task Fetcher._fetch_task has finished about aiokafka HOT 8 CLOSED

import asyncio
import logging
import time

import aiokafka


async def consume_task(consumer):
    async for msg in consumer:
        print(msg)
        await consumer.commit()


async def lag_task(producer):
    while True:
        print("sleeping")
        await asyncio.sleep(10)
        print("inducing lag")
        time.sleep(40)
        print("sending message")
        await producer.send_and_wait(
            topic='some_topic',
            value=b'a message',
        )


async def main():
    async with (
        aiokafka.AIOKafkaProducer() as producer,
        aiokafka.AIOKafkaConsumer(
            'some_topic',
            group_id='some_group',
            enable_auto_commit=False,
        ) as consumer,
    ):
        await consumer.seek_to_end()

        task1 = asyncio.create_task(consume_task(consumer))
        task2 = asyncio.create_task(lag_task(producer))

        consumer._fetcher._fetch_task.add_done_callback(lambda t: print('fetch task done'))

        await asyncio.wait([task1, task2], return_when=asyncio.FIRST_COMPLETED)
        print("something finished")


if __name__ == '__main__':
    # logging.basicConfig(level=logging.DEBUG)
    asyncio.run(main())

                    for task in self._pending_tasks:
                        # Those tasks should have proper handling for
                        # cancellation
                        if not task.done():
                            task.cancel()
                        await task

if consumer._fetcher._fetch_task.done():
   # restart app

import asyncio
from time import time

async def task_with_cleanup():
    try:
        await asyncio.sleep(1000)
    finally:
        print("Cleanup", time() - started)
        await asyncio.sleep(2)

async def worker():
    task = asyncio.create_task(task_with_cleanup())
    await asyncio.sleep(0)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("CancelledError is suppressed", time() - started)
    await asyncio.sleep(10)

async def main():
    await asyncio.wait_for(worker(), timeout=1)
    elapsed = time() - started
    assert elapsed < 2, elapsed

started = time()
asyncio.run(main())