Fix Que calls

app/integrations/kafka.py

@@ -42,7 +42,7 @@ class KafkaIntegration:
             )
 
     @staticmethod
-    def receive_disbursement_messages(topic, timeout=1.0):
+    def receive_disbursement_messages(topic, timeout):
         """
         Receive messages from a Kafka topic.
 
@@ -56,25 +56,44 @@ class KafkaIntegration:
         logger.info(
             f"Waiting for messages from topic {topic} with this timeout: {timeout}..."
         )
-
+        message =[]
         try:
             msg = consumer.poll(timeout=timeout)
+
+            logger.info(str(msg.value))
+
+            logger.info(
+                f"Received message from {msg.topic()} [{msg.partition()}] @ offset {msg.offset()}: "
+            )
+
             if msg is None:
-                logger.debug(f"No message received from topic {topic} within timeout")
+                logger.info(f"No message received from topic {topic} within timeout")
                 return None
             if msg.error():
-                logger.error(f"Consumer error: {msg.error()}")
+                logger.info(f"Consumer error: {msg.error()}")
                 raise RuntimeError(f"Consumer error: {msg.error()}")
 
             # Decode and return the message value
-            message_value = msg.value().decode("utf-8")
-            logger.debug(
+            message_value= {"value":''}
+            if msg.value() != "":
+                message_value = msg.value().decode("utf-8")
+            logger.info(
                 f"Received message from {msg.topic()} [{msg.partition()}] @ offset {msg.offset()}: {message_value}"
             )
-            message = json.loads(message_value) if message_value else None
+
+           # Invalid Json will stop this process
+            try:
+                message = json.loads(message_value) if message_value else None
+            except ValueError as e:
+                return message # Invalid JSON JUST TURN BACK HERE
+            else:
+                pass  # valid json
 
             # Call the endpoint if provided
             if message:
+                logger.info(
+                      f"Received message from {msg.topic()} [{msg.partition()}] @ offset {msg.offset()}: {message}"
+                )
                 KafkaIntegration._call_disbursement_endpoint(message)
 
             return message
@@ -83,6 +102,8 @@ class KafkaIntegration:
             logger.error(f"Error while receiving message: {e}")
             raise
 
+        #return []
+
     @staticmethod
     def close_consumer():
         """Shutdown consumer"""
@@ -101,5 +122,5 @@ class KafkaIntegration:
                 f"Successfully sent message to disbursement endpoint: {response.status_code}"
             )
         except Exception as e:
-            logger.error(f"Failed to call disbursement endpoint: {e}")
-            raise
+            logger.info(f"Failed to call disbursement endpoint: {e}")
+            #raise