fix(kafka_actions): accept list format for kafka_connect_str (DataDog#23115)

* fix(kafka_actions): accept list format for kafka_connect_str

The kafka_actions check receives kafka_connect_str from kafka_consumer
config via autodiscovery, which may be a list of broker strings. Normalize
lists to comma-separated strings (matching kafka_consumer behavior) in
both the Pydantic model validator and the config class.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* Add changelog entry for kafka_actions list fix

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: piochelepiotr

kafka_actions/changelog.d/23115.fixed

@@ -0,0 +1 @@
+Accept list format for `kafka_connect_str` when copied from kafka_consumer config via autodiscovery

kafka_actions/datadog_checks/kafka_actions/config.py

@@ -14,7 +14,10 @@ def __init__(self, instance, log):
         self.log = log
 
         self.remote_config_id = instance.get('remote_config_id')
-        self.kafka_connect_str = instance.get('kafka_connect_str')
+        kafka_connect_str = instance.get('kafka_connect_str')
+        if isinstance(kafka_connect_str, list):
+            kafka_connect_str = ','.join(str(s) for s in kafka_connect_str)
+        self.kafka_connect_str = kafka_connect_str
         self.tags = instance.get('tags', [])
 
         # Authentication fields (same pattern as kafka_consumer)

kafka_actions/datadog_checks/kafka_actions/config_models/validators.py

@@ -2,12 +2,12 @@
 # All rights reserved
 # Licensed under a 3-clause BSD style license (see LICENSE)
 
-# Here you can include additional config validators or transformers
-#
-# def initialize_instance(values, **kwargs):
-#     if 'my_option' not in values and 'my_legacy_option' in values:
-#         values['my_option'] = values['my_legacy_option']
-#     if values.get('my_number') > 10:
-#         raise ValueError('my_number max value is 10, got %s' % str(values.get('my_number')))
-#
-#     return values
+
+def initialize_instance(values, **kwargs):
+    # kafka_connect_str may be passed as a list of broker strings (e.g. from kafka_consumer config
+    # via autodiscovery). Normalize to a comma-separated string for librdkafka's bootstrap.servers.
+    kafka_connect_str = values.get('kafka_connect_str')
+    if isinstance(kafka_connect_str, list):
+        values['kafka_connect_str'] = ','.join(str(s) for s in kafka_connect_str)
+
+    return values

kafka_actions/tests/test_config_validation.py

@@ -48,6 +48,30 @@ def test_no_action_configured(self, dd_run_check):
             config = KafkaActionsConfig(instance, None)
             config.validate_config()
 
+    def test_kafka_connect_str_list(self, dd_run_check):
+        """Test that kafka_connect_str accepts a list of brokers and joins them."""
+        instance = {
+            'remote_config_id': 'test-id',
+            'kafka_connect_str': ['broker1:9092', 'broker2:9092'],
+            'produce_message': {'cluster': 'test', 'topic': 'test', 'value': 'test'},
+        }
+
+        config = KafkaActionsConfig(instance, None)
+        config.validate_config()
+        assert config.kafka_connect_str == 'broker1:9092,broker2:9092'
+
+    def test_kafka_connect_str_string(self, dd_run_check):
+        """Test that kafka_connect_str works as a plain string."""
+        instance = {
+            'remote_config_id': 'test-id',
+            'kafka_connect_str': 'broker1:9092,broker2:9092',
+            'produce_message': {'cluster': 'test', 'topic': 'test', 'value': 'test'},
+        }
+
+        config = KafkaActionsConfig(instance, None)
+        config.validate_config()
+        assert config.kafka_connect_str == 'broker1:9092,broker2:9092'
+
 
 class TestReadMessagesValidation:
     """Test read_messages action validation."""