commit b9abb09cfd89efd47bdbd8672190d2dff5982099
author lavado <glavado@whitestack.com> Fri Jul 06 21:14:00 2018 +0200
committer Gerrit Code Review <root@osm.etsi.org> Fri Jul 06 21:14:00 2018 +0200
tree 839b0d47a71555cb01ab9d2d04fbef0e8b312f53
parent 97d7df670c46be95e707d276b05c70ad61fc4ab5
parent 4515b8f6f5e3548884828eb8d96ac750e37821e3

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

policy_module/osm_policy_module/core/agent.py

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 @@
 
         # 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: