Merge pull request #20 from infn-datacloud/fix-no-kafka-brokers

Fix no kafka brokers

src/kafka_conn.py

@@ -1,7 +1,9 @@
 import json
+from logging import Logger
 from typing import Any
 
 from kafka import KafkaProducer
+from kafka.errors import NoBrokersAvailable
 
 
 class Producer:
@@ -28,3 +30,15 @@ def __init__(
     def send(self, data: dict[str, Any]):
         """Send message to kafka"""
         self.producer.send(self.topic, data)
+
+
+def get_kafka_prod(
+    *, url: str | None, topic: str | None, logger: Logger
+) -> Producer | None:
+    """Return kafka producer if broker exists."""
+    if not (url is None or topic is None):
+        try:
+            return Producer(server_url=url, topic=topic)
+        except NoBrokersAvailable:
+            logger.warning("No brokers available at %s", url)
+    return None

src/main.py

@@ -3,7 +3,7 @@
 from fed_reg.provider.schemas_extended import ProviderCreateExtended
 
 from src.config import get_settings
-from src.kafka_conn import Producer
+from src.kafka_conn import get_kafka_prod
 from src.logger import create_logger
 from src.parser import parser
 from src.providers.core import ProviderThread
@@ -27,11 +27,9 @@ def main(log_level: str) -> None:
     settings = get_settings()
 
     # Create kafka producer if needed
-    kafka_prod = None
-    if not (settings.KAFKA_SERVER_URL is None or settings.KAFKA_TOPIC is None):
-        kafka_prod = Producer(
-            server_url=settings.KAFKA_SERVER_URL, topic=settings.KAFKA_TOPIC
-        )
+    kafka_prod = get_kafka_prod(
+        url=settings.KAFKA_SERVER_URL, topic=settings.KAFKA_TOPIC, logger=logger
+    )
 
     # Read all yaml files containing providers configurations.
     yaml_files = get_conf_files(settings=settings, logger=logger)