projects
/
osm
/
MON.git
/ blobdiff
commit
grep
author
committer
pickaxe
?
search:
re
summary
|
shortlog
|
log
|
commit
|
commitdiff
|
tree
raw
|
inline
| side by side
Added LICENSE file to root folder
[osm/MON.git]
/
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
(file)
--- 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)
# 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,
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:
consumer.subscribe(['lcm_pm', 'alarm_response'])
for message in consumer: