ISSUE #336

* Minor improvement

Author: Sergio García Prado

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

@@ -169,8 +169,10 @@ def admin_client(self):
     async def _receive(self) -> BrokerMessage:
         try:
             record = await self.client.getone()
-        except ConsumerStoppedError:
-            raise StopAsyncIteration
+        except ConsumerStoppedError as exc:
+            if self.already_destroyed:
+                raise StopAsyncIteration
+            raise exc
         bytes_ = record.value
         message = BrokerMessage.from_avro_bytes(bytes_)
         return message

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

@@ -207,12 +207,20 @@ async def test_receive(self):
             self.assertEqual(messages[0], await subscriber.receive())
             self.assertEqual(messages[1], await subscriber.receive())
 
+    async def test_receive_already_stopped_raises(self):
+        subscriber = KafkaBrokerSubscriber.from_config(CONFIG_FILE_PATH, topics={"foo", "bar"})
+        get_mock = AsyncMock(side_effect=ConsumerStoppedError)
+        subscriber.client.getone = get_mock
+
+        with self.assertRaises(StopAsyncIteration):
+            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):
+            with self.assertRaises(ConsumerStoppedError):
                 await subscriber.receive()