progress on kafka

app/integrations/kafka.py

@@ -1,4 +1,4 @@
-from confluent_kafka import Consumer
+from confluent_kafka import Consumer, Producer
 import json
 import logging
 from app.config import settings
@@ -31,7 +31,6 @@ class KafkaIntegration:
     def delivery_report(err, msg):
         """Called once for each message produced"""
         if err is not None:
-
             logger.error(f"Message delivery failed: {err}")
             raise RuntimeError(f"Message delivery failed: {err}")
 
@@ -41,7 +40,7 @@ class KafkaIntegration:
             )
 
     @staticmethod
-    def receive_messages(topic, timeout=1.0):
+    def receive_disbursement_messages(topic, timeout=1.0):
         """
         Receive messages from a Kafka topic.
 
@@ -66,7 +65,13 @@ class KafkaIntegration:
             logger.debug(
                 f"Received message from {msg.topic()} [{msg.partition()}] @ offset {msg.offset()}: {message_value}"
             )
-            return json.loads(message_value) if message_value else None
+            message = json.loads(message_value) if message_value else None
+
+            # Call the endpoint if provided
+            if message:
+                KafkaIntegration._call_disbursement_endpoint(message)
+
+            return message
 
         except Exception as e:
             logger.error(f"Error while receiving message: {e}")
@@ -76,29 +81,9 @@ class KafkaIntegration:
     def close_consumer():
         """Shutdown consumer"""
         consumer = KafkaIntegration._get_consumer()
-        consumer.clonse()
+        consumer.close()
         logger.info("Kafka consumer closed")
 
-    @staticmethod
-    def is_kafka_running():
-        """
-        Check if Kafka broker is reachable by attempting to fetch metadata.
-        :return: True if Kafka is running, False otherwise
-        """
-        try:
-            producer = KafkaIntegration._get_producer()
-            # Attempt to fetch metadata for a dummy topic
-            metadata = producer.list_topics(timeout=5)
-            if metadata.brokers:  # If brokers are returned, Kafka is running
-                logger.info(f"Kafka broker at {settings.KAFKA_BROKER} is running")
-                return True
-            else:
-                logger.warning(f"No brokers found for {settings.KAFKA_BROKER}")
-                return False
-        except Exception as e:
-            logger.error(f"Kafka broker check failed: {e}")
-            return False
-
     @staticmethod
     def _call_disbursement_endpoint(
         message, endpoint_url="http://localhost:8000/loans/disbursement"