From: Benjamin Diaz <bdiaz@whitestack.com>
Date: Fri, 6 Jul 2018 15:03:35 +0000 (-0300)
Subject: [PM] Adds group-id to Kafka consumer in agent
X-Git-Tag: BUILD_v4.0.1_1~2^2
X-Git-Url: https://osm.etsi.org/gitweb/?a=commitdiff_plain;h=4515b8f6f5e3548884828eb8d96ac750e37821e3;p=osm%2FMON.git

[PM] Adds group-id to Kafka consumer in agent

Fixes CPU spikes during OSM startup

Signed-off-by: Benjamin Diaz <bdiaz@whitestack.com>
---

diff --git a/policy_module/osm_policy_module/core/agent.py b/policy_module/osm_policy_module/core/agent.py
index 3a3b20c..cdd5dfc 100644
--- a/policy_module/osm_policy_module/core/agent.py
+++ b/policy_module/osm_policy_module/core/agent.py
@@ -45,11 +45,10 @@ class PolicyModuleAgent:
 
         # Initialize Kafka consumer
         log.info("Connecting to Kafka server at %s", kafka_server)
-        # TODO: Add logic to handle deduplication of messages when using group_id.
-        # See: https://stackoverflow.com/a/29836412
         consumer = KafkaConsumer(bootstrap_servers=kafka_server,
                                  key_deserializer=bytes.decode,
-                                 value_deserializer=bytes.decode)
+                                 value_deserializer=bytes.decode,
+                                 group_id="pm-consumer")
         consumer.subscribe(['lcm_pm', 'alarm_response'])
 
         for message in consumer: