From 264c655ccf55097ffee7afdc6ac612da242b6aaf Mon Sep 17 00:00:00 2001 From: Atul Agarwal Date: Tue, 15 Jun 2021 06:53:27 +0000 Subject: [PATCH] Resolved Bug 1569 - Unable to subscribe to Kafka topics Change-Id: I2cde18ff8169ed2e8cecb531fb3814d53eb77b04 Signed-off-by: Atul Agarwal --- osm_mon/dashboarder/dashboarder.py | 19 ++++++++++++------- osm_mon/server/server.py | 20 +++++++++++++------- 2 files changed, 25 insertions(+), 14 deletions(-) diff --git a/osm_mon/dashboarder/dashboarder.py b/osm_mon/dashboarder/dashboarder.py index bed157a..a9c8a57 100644 --- a/osm_mon/dashboarder/dashboarder.py +++ b/osm_mon/dashboarder/dashboarder.py @@ -47,14 +47,19 @@ class Dashboarder: def run(self): self.loop.run_until_complete(self.start()) - async def start(self): + async def start(self, wait_time=5): topics = ["users", "project"] - try: - await self.msg_bus.aioread(topics, self._user_msg) - except Exception as e: - # Failed to subscribe to kafka topics - log.error("Error when subscribing to topics %s", str(topics)) - log.exception("Exception %s", str(e)) + while True: + try: + await self.msg_bus.aioread(topics, self._user_msg) + break + except Exception as e: + # Failed to subscribe to kafka topics + log.error("Error when subscribing to topics %s", str(topics)) + log.exception("Exception %s", str(e)) + # Wait for some time for kaka to stabilize and then reattempt to subscribe again + time.sleep(wait_time) + log.info("Retrying to subscribe the kafka topic(s) %s", str(topics)) async def _user_msg(self, topic, key, values): log.debug( diff --git a/osm_mon/server/server.py b/osm_mon/server/server.py index 62721ff..6df6ed0 100755 --- a/osm_mon/server/server.py +++ b/osm_mon/server/server.py @@ -26,6 +26,7 @@ MON component in charge of CRUD operations for vim_accounts and alarms. It uses import asyncio import json import logging +import time from osm_mon.core.config import Config from osm_mon.core.message_bus_client import MessageBusClient @@ -48,14 +49,19 @@ class Server: def run(self): self.loop.run_until_complete(self.start()) - async def start(self): + async def start(self, wait_time=5): topics = ["alarm_request"] - try: - await self.msg_bus.aioread(topics, self._process_msg) - except Exception as e: - # Failed to subscribe to kafka topic - log.exception("Error when subscribing to topics %s", str(topics)) - raise e + while True: + try: + await self.msg_bus.aioread(topics, self._process_msg) + break + except Exception as e: + # Failed to subscribe to kafka topic + log.exception("Error when subscribing to topics %s", str(topics)) + log.exception("Exception %s", str(e)) + # Wait for some time for kaka to stabilize and then reattempt to subscribe again + time.sleep(wait_time) + log.info("Retrying to subscribe the kafka topic(s) %s", str(topics)) async def _process_msg(self, topic, key, values): log.info("Message arrived: %s", values) -- 2.25.1