BREAKING CHANGE: Replacing host & port by bootstrap_servers

feat: Add client_id to consumer

Author: sevenup30

extensions/eda/plugins/event_source/kafka.py

@@ -13,15 +13,13 @@
 description:
   - An ansible-rulebook event source plugin for receiving events via a kafka topic.
 options:
-  host:
+  bootstrap_servers:
     description:
-      - The host where the kafka topic is hosted.
-    type: str
-    required: true
-  port:
-    description:
-      - The port where the kafka server is listening.
-    type: str
+      - A host[:port] string (or 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: str,list(str)
     required: true
   cafile:
     description:
@@ -87,6 +85,14 @@
       - A kafka group id.
     type: str
     default: null
+  client_id:
+    description:
+      - client_id (str) – a name for this client.
+        This string is passed in each request to servers and can be used to identify specific server-side log entries
+        that correspond to this client.
+        Also submitted to GroupCoordinator for logging with respect to consumer group administration
+    type: str
+    default: null
   offset:
     description:
       - Where to automatically reset the offset.
@@ -125,15 +131,15 @@
 
 EXAMPLES = r"""
 - ansible.eda.kafka:
-    host: "localhost"
-    port: "9092"
+    bootstrap_servers: broker-1:9092,broker-2:9093:broker-3:9094
     check_hostname: true
     verify_mode: "CERT_OPTIONAL"
     encoding: "utf-8"
     topics:
       - "demo"
       - "demo2"
     group_id: "test"
+    client_id: "client-01"
     offset: "earliest"
     security_protocol: "SASL_PLAINTEXT"
     sasl_mechanism: "GSSAPI"
@@ -161,15 +167,15 @@ async def main(  # pylint: disable=R0914
     if topic:
         topics = [topic]
 
-    host = args.get("host")
-    port = args.get("port")
+    bootstrap_servers = args.get("bootstrap_servers")
     cafile = args.get("cafile")
     certfile = args.get("certfile")
     keyfile = args.get("keyfile")
     password = args.get("password")
     check_hostname = args.get("check_hostname", True)
     verify_mode = args.get("verify_mode", "CERT_REQUIRED")
     group_id = args.get("group_id")
+    client_id = args.get("client_id", "")
     offset = args.get("offset", "latest")
     encoding = args.get("encoding", "utf-8")
     security_protocol = args.get("security_protocol", "PLAINTEXT")
@@ -202,8 +208,9 @@ async def main(  # pylint: disable=R0914
         ssl_context.verify_mode = verify_mode
 
     kafka_consumer = AIOKafkaConsumer(
-        bootstrap_servers=f"{host}:{port}",
+        bootstrap_servers=bootstrap_servers,
         group_id=group_id,
+        client_id=client_id,
         enable_auto_commit=True,
         max_poll_records=1,
         auto_offset_reset=offset,

tests/integration/event_source_kafka/test_kafka_rules_headers.yml

@@ -3,8 +3,7 @@
   sources:
     - ansible.eda.kafka:
         topic: kafka-events-plaintext
-        host: localhost
-        port: 9092
+        bootstrap_servers: localhost:9092
         offset: earliest
         encoding: ascii
   rules:

tests/integration/event_source_kafka/test_kafka_rules_multiple_topics.yml

@@ -5,8 +5,7 @@
         topics:
           - kafka-events-topic1
           - kafka-events-topic2
-        host: localhost
-        port: 9092
+        bootstrap_servers: localhost:9092
         offset: earliest
         encoding: ascii
   rules:

tests/integration/event_source_kafka/test_kafka_rules_pattern_topics.yml

@@ -3,8 +3,7 @@
   sources:
     - ansible.eda.kafka:
         topic_pattern: "kafka-pattern-.*"
-        host: localhost
-        port: 9092
+        bootstrap_servers: localhost:9092
         offset: earliest
         encoding: ascii
   rules:

tests/integration/event_source_kafka/test_kafka_rules_plaintext.yml

@@ -3,8 +3,7 @@
   sources:
     - ansible.eda.kafka:
         topic: kafka-events-plaintext
-        host: localhost
-        port: 9092
+        bootstrap_servers: localhost:9092
         offset: earliest
         encoding: ascii
   rules:

tests/integration/event_source_kafka/test_kafka_rules_sasl_plaintext.yml

@@ -3,8 +3,7 @@
   sources:
     - ansible.eda.kafka:
         topic: kafka-events-sasl-plaintext
-        host: localhost
-        port: 9094
+        bootstrap_servers: localhost:9094
         offset: earliest
         encoding: ascii
         security_protocol: SASL_PLAINTEXT

tests/integration/event_source_kafka/test_kafka_rules_sasl_ssl.yml

@@ -3,8 +3,7 @@
   sources:
     - ansible.eda.kafka:
         topic: kafka-events-sasl-ssl
-        host: localhost
-        port: 9095
+        bootstrap_servers: localhost:9095
         offset: earliest
         encoding: ascii
         security_protocol: SASL_SSL

tests/integration/event_source_kafka/test_kafka_rules_ssl.yml

@@ -3,8 +3,7 @@
   sources:
     - ansible.eda.kafka:
         topic: kafka-events-ssl
-        host: localhost
-        port: 9093
+        bootstrap_servers: localhost:9093
         offset: earliest
         encoding: ascii
         security_protocol: SSL

tests/unit/event_source/test_kafka.py

@@ -66,8 +66,7 @@ def test_receive_from_kafka_place_in_queue(
                 myqueue,
                 {
                     topic_type: topic_value,
-                    "host": "localhost",
-                    "port": "9092",
+                    "bootstrap_servers": "localhost:9092",
                     "group_id": "test",
                 },
             )