Place the key in the right place

yuptoo/lib/produce.py

@@ -21,7 +21,7 @@ def init_producer():
     return producer
 
 
-def send_message(kafka_topic, msg, request_obj=None):
+def send_message(kafka_topic, msg, kafka_message_key, request_obj=None):
 
     def delivery_report(err, msg=None, is_msg_for_hbi=False):
         nonlocal request_obj
@@ -37,9 +37,10 @@ def delivery_report(err, msg=None, is_msg_for_hbi=False):
 
     try:
         bytes = json.dumps(msg, ensure_ascii=False).encode("utf-8")
+        key = kafka_message_key.encode("utf-8")
         if kafka_topic == UPLOAD_TOPIC:
-            org_id = request_obj.get('org_id') if request_obj else None
-            key = org_id.encode("utf-8") if org_id else None
+            if not key:
+                raise KafkaException ("Org_id as a key in Kafka messages destined for hbi")
             producer.produce(kafka_topic, bytes, key, callback=partial(delivery_report, is_msg_for_hbi=True))
         else:
             producer.produce(kafka_topic, bytes, callback=delivery_report)

yuptoo/processor/report_processor.py

@@ -90,12 +90,11 @@ def upload_to_host_inventory_via_kafka(host, request_obj):
     try:
         upload_msg = {
             'operation': 'add_host',
-            'key': host.org_id,
             'data': host,
             'platform_metadata': {'request_id': host['system_unique_id'],
                                   'b64_identity': request_obj['b64_identity']}
         }
-        send_message(UPLOAD_TOPIC, upload_msg, request_obj)
+        send_message(UPLOAD_TOPIC, upload_msg, host.org_id, request_obj)
     except Exception as err:
         LOG.error(f"The following error occurred: {err}")