commit 9ea2154491ecd2324cde998ce595c9430a14516b
author Zack Williams <zdw@opennetworking.org> Fri Oct 26 10:36:31 2018 -0700
committer khenaidoo Nursimulu <knursimu@ciena.com> Sun Nov 04 22:46:14 2018 +0000
tree 964d00670cdb36c77fed274637c40d1e5227068b
parent b4868197f4294e1f6fb87822f2e323378a563241

Add the ability to set the Kafka key in KafkaProxy.send_message()

Change-Id: Ib10e777b4ae3e8a4ce469a7d8e74bf5b47de4eda

voltha/northbound/kafka/kafka_proxy.py

diff --git a/voltha/northbound/kafka/kafka_proxy.py b/voltha/northbound/kafka/kafka_proxy.py
index 6e80252..f0c58c7 100644
--- a/voltha/northbound/kafka/kafka_proxy.py
+++ b/voltha/northbound/kafka/kafka_proxy.py
@@ -143,7 +143,7 @@
             return
 
     @inlineCallbacks
-    def send_message(self, topic, msg):
+    def send_message(self, topic, msg, key=None):
         assert topic is not None
         assert msg is not None
 
@@ -164,7 +164,7 @@
                 msgs = [msg]
 
                 if self.kproducer is not None and self.event_bus_publisher and self.faulty is False:
-                    d =  deferToThread(self.kproducer.produce, topic, msg)
+                    d =  deferToThread(self.kproducer.produce, topic, msg, key)
                     yield d
                     log.debug('sent-kafka-msg', topic=topic, msg=msg)
                 else: