X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=policy_module%2Fosm_policy_module%2Fcore%2Fagent.py;h=cdd5dfc9eb8b8b24b8307a20fdfa39daa5b0015c;hb=a4ffa936b4f8a2bfcd1b64541e00c93aac449a56;hp=aa3f3ff3d2db4a20bbea2e4463cd508b0e5108c5;hpb=e2fe4582e9569ea8146e945f278179126821ab76;p=osm%2FMON.git

diff --git a/policy_module/osm_policy_module/core/agent.py b/policy_module/osm_policy_module/core/agent.py
index aa3f3ff..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:
@@ -113,7 +112,7 @@ class PolicyModuleAgent:
                         log.info("Sending scaling action message for ns: %s", alarm_id)
                         lcm_client.scale(alarm.scaling_record.nsr_id, alarm.scaling_record.name, alarm.action)
                     except ScalingAlarm.DoesNotExist:
-                        log.info("There is no action configured for alarm %.", alarm_id)
+                        log.info("There is no action configured for alarm %s.", alarm_id)
             except Exception:
                 log.exception("Error consuming message: ")