Adds OSMMON_VCA_USER and adds timeout and max.poll.interval to collector

[osm/MON.git] / osm_mon / core / message_bus / common_consumer.py

diff --git a/osm_mon/core/message_bus/common_consumer.py b/osm_mon/core/message_bus/common_consumer.py
index 85e679f..e32fa2b 100755 (executable)
--- a/osm_mon/core/message_bus/common_consumer.py
+++ b/osm_mon/core/message_bus/common_consumer.py
@@ -22,6 +22,7 @@
 import json
 import logging
 import sys
+import time
 from json import JSONDecodeError
 
 import six
@@ -51,10 +52,6 @@ log = logging.getLogger(__name__)
 
 kafka_logger = logging.getLogger('kafka')
 kafka_logger.setLevel(logging.getLevelName(cfg.OSMMON_KAFKA_LOG_LEVEL))
-kafka_formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
-kafka_handler = logging.StreamHandler(sys.stdout)
-kafka_handler.setFormatter(kafka_formatter)
-kafka_logger.addHandler(kafka_handler)
 
 
 class CommonConsumer:
@@ -106,6 +103,7 @@ class CommonConsumer:
                     log.exception("Exception: ")
                     return
                 retries = retries + 1
+                time.sleep(2)
 
         log.info("Listening for messages...")
         for message in common_consumer:
@@ -196,7 +194,7 @@ class CommonConsumer:
         key = key.replace('request', 'response')
         producer = Producer()
         producer.send(topic=topic, key=key, value=json.dumps(msg))
-        producer.flush()
+        producer.flush(timeout=5)
         producer.close()