ISSUE #?

* Raise a `StopAsyncIteration` when consumer is stopped.

Author: Sergio García Prado

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

@@ -166,10 +166,9 @@ def admin_client(self):
     async def _receive(self) -> BrokerMessage:
         try:
             record = await self.client.getone()
-        except ConsumerStoppedError as exc:
-            if self.already_destroyed:
-                raise StopAsyncIteration
-            raise exc
+        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

@@ -201,20 +201,12 @@ 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(ConsumerStoppedError):
+            with self.assertRaises(StopAsyncIteration):
                 await subscriber.receive()