Uncaught ConsumerStoppedError exceptions on closed streams

[JeroennC]

Describe the bug
Streams running with the UDF approach are not silently shutting down when the consumers are stopped. The ConsumerStoppedError raised is never caught. In our application this is causing sentry errors while this is a graceful shutdown.

To Reproduce
Run a UDF consumer stream and shut down the application.

Expected behavior
I would expect stream.start() to suppress the ConsumerStoppedError.

Logs
Task exception was never retrieved
future: <Task finished name='Task-10' coro=<Stream.start() done, defined at /app/.venv/lib/python3.11/site-packages/kstreams/streams.py:334> exception=ConsumerStoppedError()>

ConsumerStoppedError: null
File "kstreams/streams.py", line 361, in start
await self.func_wrapper_with_typing()
File "kstreams/streams.py", line 367, in func_wrapper_with_typing
cr = await self.getone()
File "kstreams/streams.py", line 277, in getone
consumer_record: ConsumerRecord = await self.consumer.getone() # type: ignore
File "aiokafka/consumer/consumer.py", line 1160, in getone
msg = await self._fetcher.next_record(partitions)
File "aiokafka/consumer/fetcher.py", line 1086, in next_record
raise ConsumerStoppedError()

[marcosschroh]

closed by #238