handle kafka exceptions

eventify/drivers/kafka.py

@@ -122,18 +122,23 @@ async def onJoin(self):
                     # Used with config.json defined topics
                     if self.subscribed_topics is not None:
                         consumer = AIOKafkaConsumer(
-                            ', '.join(self.subscribed_topics),
                             bootstrap_servers=self.transport_host,
                             loop=loop,
                             group_id='my-group'
                         )
                         await consumer.start()
+
+                        # Subscribe to all topics
+                        for topic in self.subscribed_topics:
+                            consumer.subscribe(topic)
+
                         try:
                             async for msg in consumer:
                                 value = msg.value.decode()
                                 await handler_instance.on_event(value)
-                        finally:
-                            await consumer.stop()
+                        except Exception as error:
+                            self.log.error("Consumer error. %s", error)
+                            await asyncio.sleep(0)
 
 
             if hasattr(handler_instance, 'worker'):

setup.py

@@ -4,12 +4,12 @@
 setup(
     name='eventify',
     packages=find_packages(),
-    version='0.5.0',
+    version='0.5.1',
     description='Event Driven Asynchronous Framework',
     author='Matthew Harris',
     author_email='matt@x-qa.com',
     url='https://github.com/eventifyio/eventify',
-    download_url='https://github.com/eventifyio/eventify/raw/master/dist/eventify-0.5.0.tar.gz',
+    download_url='https://github.com/eventifyio/eventify/raw/master/dist/eventify-0.5.1.tar.gz',
     keywords=['event', 'event-driven', 'async',
               'framework', 'producer', 'consumer'],
     classifiers=[],