Merge pull request #340 from minos-framework/issue-336-aiokafka-rare-errors

#336 - Fix `AioKafkaConsumer` strange errors

Author: Sergio García Prado

packages/plugins/minos-broker-kafka/minos/plugins/kafka/subscriber.py

@@ -22,6 +22,7 @@
 
 from aiokafka import (
     AIOKafkaConsumer,
+    ConsumerStoppedError,
 )
 from cached_property import (
     cached_property,
@@ -166,7 +167,10 @@ def admin_client(self):
         return KafkaAdminClient(bootstrap_servers=f"{self.host}:{self.port}")
 
     async def _receive(self) -> BrokerMessage:
-        record = await self.client.getone()
+        try:
+            record = await self.client.getone()
+        except ConsumerStoppedError:
+            raise StopAsyncIteration
         bytes_ = record.value
         message = BrokerMessage.from_avro_bytes(bytes_)
         return message

packages/plugins/minos-broker-kafka/tests/test_kafka/test_subscriber.py

@@ -10,6 +10,7 @@
 
 from aiokafka import (
     AIOKafkaConsumer,
+    ConsumerStoppedError,
 )
 from kafka import (
     KafkaAdminClient,
@@ -206,6 +207,14 @@ async def test_receive(self):
             self.assertEqual(messages[0], await subscriber.receive())
             self.assertEqual(messages[1], await subscriber.receive())
 
+    async def test_receive_stopped(self):
+        async with KafkaBrokerSubscriber.from_config(CONFIG_FILE_PATH, topics={"foo", "bar"}) as subscriber:
+            get_mock = AsyncMock(side_effect=ConsumerStoppedError)
+            subscriber.client.getone = get_mock
+
+            with self.assertRaises(StopAsyncIteration):
+                await subscriber.receive()
+
 
 class TestKafkaBrokerSubscriberBuilder(unittest.TestCase):
     def setUp(self) -> None: