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

[AxTheB]

When we replaced the fetch task with fresh one, it did not receive new messages (there were messages in assigned partition) and got finished too, without logging anything. I see only one way for this task to get to finished state, and that is by swallowing the CancelledError.
If the error got propagated, it would allow the app to restart and quickly resume consuming messages.

[ods]

The only place where _fetch_task is cancelled is the Fetcher.close() method. Or do you see other options? Also, for this exception to be propagated, you have to await the task, which is also done in close() method only.

[ods]

Hmm, there is other option to get CancelledError here:

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

[pavelschon]

Can you confirm this is a bug in aiokafka?

On our side, we had to implement a workaround

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