Update main.py

main.py

@@ -8,6 +8,7 @@ from typing import List, Dict, Union
 import time
 from kafka import KafkaConsumer, KafkaProducer
 import asyncio
+import json
 
 
 app = FastAPI()
@@ -31,11 +32,12 @@ async def kafka_consumer():
             obj_id = data.get('obj_id')
             raw_data = {"text":data.get('text', str()), 'star_rating':data.get('rating', 5), "skip":False}
             processed_data, has_sentiments = process_single_comment(raw_data)
+            print(f"Processed message with id : {obj_id}")
             processed_data.pop('color_map', None)
             processed_data_dict = {"processed_data":processed_data, "has_sentiments":has_sentiments,"obj_id":obj_id}
             producer = KafkaProducer(bootstrap_servers=bootstrap_server)
             msg = json.dumps(processed_data_dict)
-            producer.send(self.topic, message.encode('utf-8'))
+            producer.send(producer_topic, msg.encode('utf-8'))
             producer.flush()
     finally:
         consumer.close()