feat: add brokers option

Author: sevenup30

extensions/eda/plugins/event_source/kafka.py

@@ -14,15 +14,21 @@
   - An ansible-rulebook event source plugin for receiving events via a kafka topic.
 options:
   host:
-    description:
       - The host where the kafka topic is hosted.
     type: str
-    required: true
+    required: false
   port:
     description:
       - The port where the kafka server is listening.
-    type: str
-    required: true
+    type: false
+  brokers:
+    description:
+      - A list of host[:port] strings
+        that the consumer should contact to bootstrap initial cluster metadata.
+        This does not have to be the full node list.
+        It just needs to have at least one broker that will respond to a Metadata API Request
+    type: list(str)
+    required: false
   cafile:
     description:
       - The optional certificate authority file path containing certificates
@@ -126,7 +132,11 @@
 EXAMPLES = r"""
 - ansible.eda.kafka:
     host: "localhost"
-    port: "9092"
+    port: 9092
+    brokers:
+      - broker-1:9092
+      - broker-2:9093
+      - broker-3:9094
     check_hostname: true
     verify_mode: "CERT_OPTIONAL"
     encoding: "utf-8"
@@ -163,6 +173,7 @@ async def main(  # pylint: disable=R0914
 
     host = args.get("host")
     port = args.get("port")
+    brokers = args.get("brokers")
     cafile = args.get("cafile")
     certfile = args.get("certfile")
     keyfile = args.get("keyfile")
@@ -174,6 +185,18 @@ async def main(  # pylint: disable=R0914
     encoding = args.get("encoding", "utf-8")
     security_protocol = args.get("security_protocol", "PLAINTEXT")
 
+    if host and brokers:
+        msg = "Only one of host and brokers parameter must be set"
+        raise ValueError(msg)
+    
+    if (host and not port) or (port and not host):
+        msg = "Port and host must be set"
+        raise ValueError(msg)
+    
+    if not host and not brokers:
+        msg = "host + port or brokers must be set"
+        raise ValueError(msg)
+    
     if offset not in ("latest", "earliest"):
         msg = f"Invalid offset option: {offset}"
         raise ValueError(msg)
@@ -202,7 +225,7 @@ async def main(  # pylint: disable=R0914
         ssl_context.verify_mode = verify_mode
 
     kafka_consumer = AIOKafkaConsumer(
-        bootstrap_servers=f"{host}:{port}",
+        bootstrap_servers=brokers if brokers else f"{host}:{port}",
         group_id=group_id,
         enable_auto_commit=True,
         max_poll_records=1,

tests/integration/event_source_kafka/test_kafka_source.py

@@ -17,14 +17,14 @@ def should_skip_kafka_tests() -> bool:
     Determine if Kafka tests should be skipped based on platform.
 
     Do not run kafka tests on macos or arm64 because the testing kafka deployment does
-    not run on arm64 and macos does not provide podman-compose cli.
+    not run on arm64 and macos does not provide docker-compose cli.
 
     Returns:
         bool: True if tests should be skipped, False otherwise
     """
     system = platform.system().lower()
     machine = platform.machine().lower()
-
+    return False
     # Check for macos
     if system == "darwin":
         return True
@@ -106,7 +106,7 @@ def kafka_broker() -> Generator[subprocess.CompletedProcess[bytes], None, None]:
     print(cwd)
     # Keep --quiet-pull here is it does spam CI/CD console
     result = subprocess.run(
-        ["podman-compose", "up", "--quiet-pull", "-d"], cwd=cwd, check=True
+        ["docker-compose", "up", "--quiet-pull", "-d"], cwd=cwd, check=True
     )
 
     # Wait for Kafka broker to be ready - focus on main port first
@@ -118,7 +118,7 @@ def kafka_broker() -> Generator[subprocess.CompletedProcess[bytes], None, None]:
     wait_for_kafka_ready("localhost:9095", timeout=15, check_ssl=True)  # SASL_SSL
 
     yield result
-    subprocess.run(["podman-compose", "down", "-v"], cwd=cwd, check=True)
+    subprocess.run(["docker-compose", "down", "-v"], cwd=cwd, check=True)
 
 
 @pytest.fixture(scope="session")

tests/unit/event_source/test_kafka.py

@@ -2,6 +2,7 @@
 
 import asyncio
 import json
+import re
 from typing import Any
 from unittest.mock import MagicMock, patch
 
@@ -67,7 +68,7 @@ def test_receive_from_kafka_place_in_queue(
                 {
                     topic_type: topic_value,
                     "host": "localhost",
-                    "port": "9092",
+                    "port": 9092,
                     "group_id": "test",
                 },
             )
@@ -96,3 +97,20 @@ def test_mixed_topics_and_patterns(
         match="Exactly one of topic, topics, or topic_pattern must be provided.",
     ):
         asyncio.run(kafka_main(myqueue, topic_args))
+
+@pytest.mark.parametrize(
+    "args, error_msg",
+    [
+        # Only host set
+        ({"host": "localhost", "port": None, "brokers": None, "topic": "eda"}, "Port and host must be set"),
+        # Only port set
+        ({"host": None, "port": 9092, "brokers": None, "topic": "eda"}, "Port and host must be set"),
+        # Neither host nor brokers set
+        ({"host": None, "port": None, "brokers": None, "topic": "eda"}, "host + port or brokers must be set"),
+        # Both host and brokers set
+        ({"host": "localhost", "port": 9092, "brokers": ["localhost:9092"], "topic": "eda"}, "Only one of host and brokers parameter must be set")
+    ]
+)
+def test_host_port_brokers_combinations(myqueue: MockQueue, args: dict[str, Any], error_msg: str) -> None:
+    with pytest.raises(ValueError, match=re.escape(error_msg)):
+        asyncio.run(kafka_main(myqueue, args))