encode
Kafka backend can't unsubscribe from individual channel
Currently, unsubscribing from one Kafka channel unsubscribes from all channels. Based on the aiokafka docs I'm guessing we want to do
self._consumer_channels.remove(channel)
self._consumer.subscribe(topics=self._consumer_channels)
Also, when we call AIOKafkaConsumer we might want to add auto_offset_reset="latest" based on https://aiokafka.readthedocs.io/en/stable/consumer.html#controlling-the-consumer-s-position. Even then, when we change the topics we're subscribed to it's not obvious to me that we won't miss events or process events multiple times.
Currently, unsubscribing from one Kafka channel unsubscribes from all channels. Based on the aiokafka docs I'm guessing we want to do
self._consumer_channels.remove(channel) self._consumer.subscribe(topics=self._consumer_channels)
When _counsumer_channels will be empty it raise error, so:
async def unsubscribe(self, channel: str) -> None:
self._consumer_channels.remove(channel)
if self._consumer_channels:
self._consumer.subscribe(topics=list(self._consumer_channels))
else:
self._consumer.unsubscribe()