kafka_consumer: add broker list to cluster monitoring heartbeat (DataDog#23898)

* Add broker list to kafka_consumer cluster monitoring heartbeat

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* Add changelog entry for PR DataDog#23898

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* Address review comments on heartbeat broker list

- Extract _get_broker_list() helper to separate topology-transform from payload-assembly
- Use broker_meta.id instead of dict key; cast to str to match broker config event convention
- Add tests for heartbeat brokers payload (populated and empty cases)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* Fix _cluster_metadata not set on create_autospec mock in tests

seed_mock_kafka_client uses create_autospec(KafkaClient), which does not
expose instance attributes set in __init__. Setting _cluster_metadata
directly on the mock fixture makes it available to _get_broker_list().

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: piochelepiotr

kafka_consumer/changelog.d/23898.added

@@ -0,0 +1 @@
+Add broker list to the cluster monitoring heartbeat payload.

kafka_consumer/datadog_checks/kafka_consumer/kafka_consumer.py

@@ -131,6 +131,15 @@ def check(self, _):
     def count_consumer_contexts(self, consumer_offsets):
         return sum(len(offsets) for offsets in consumer_offsets.values())
 
+    def _get_broker_list(self) -> list[dict]:
+        cluster_metadata = self.client._cluster_metadata
+        if not (cluster_metadata and hasattr(cluster_metadata, 'brokers')):
+            return []
+        return [
+            {'id': str(broker_meta.id), 'host': broker_meta.host, 'port': broker_meta.port}
+            for broker_meta in cluster_metadata.brokers.values()
+        ]
+
     def _send_cluster_monitoring_heartbeat(self, total_contexts: int, cluster_id: str) -> None:
         payload = {
             'collection_timestamp': int(time() * 1000),
@@ -139,6 +148,7 @@ def _send_cluster_monitoring_heartbeat(self, total_contexts: int, cluster_id: st
             'contexts': total_contexts,
             'contexts_limit': self._context_limit,
             'bootstrap_servers': self.config._kafka_connect_str,
+            'brokers': self._get_broker_list(),
         }
         if self.config._kafka_cluster_id_override:
             payload['original_kafka_cluster_id'] = self.config._auto_detected_cluster_id

kafka_consumer/tests/test_cluster_metadata.py

@@ -146,6 +146,7 @@ def mock_describe_configs(resources):
 
     # Set kafka_client as an attribute (not a property mock)
     client.kafka_client = mock_admin_client
+    client._cluster_metadata = metadata
     client.get_topic_partitions.return_value = {'test-topic': [0, 1]}
 
     def mock_offsets_for_times(partitions, offset=-1):
@@ -1419,3 +1420,41 @@ def test_partition_out_of_sync_broker_id_tag(
         'kafka.partition.offline',
     ):
         aggregator.assert_metric(metric, tags=expected_tags)
+
+
+def test_heartbeat_brokers_populated(check):
+    """Heartbeat payload includes the broker list when metadata is available."""
+    instance = {'kafka_connect_str': 'localhost:9092', 'enable_cluster_monitoring': True}
+    kafka_consumer_check = check(instance)
+    mock_kafka_client = seed_mock_kafka_client()
+    kafka_consumer_check.client = mock_kafka_client
+    kafka_consumer_check.event_platform_event = mock.Mock()
+
+    kafka_consumer_check._send_cluster_monitoring_heartbeat(total_contexts=5, cluster_id='test-cluster-id')
+
+    calls = kafka_consumer_check.event_platform_event.call_args_list
+    hb_events = [json.loads(c[0][0]) for c in calls if c[0][1] == 'data-streams-message']
+    hb_events = [e for e in hb_events if e.get('config_type') == 'heartbeat']
+    assert len(hb_events) == 1
+    assert hb_events[0]['brokers'] == [
+        {'id': '1', 'host': 'broker1', 'port': 9092},
+        {'id': '2', 'host': 'broker2', 'port': 9092},
+    ]
+
+
+def test_heartbeat_brokers_empty_when_no_metadata(check):
+    """Heartbeat payload has an empty broker list when _cluster_metadata is None."""
+    instance = {'kafka_connect_str': 'localhost:9092', 'enable_cluster_monitoring': True}
+    kafka_consumer_check = check(instance)
+    mock_kafka_client = seed_mock_kafka_client()
+    mock_kafka_client._cluster_metadata = None
+    kafka_consumer_check.client = mock_kafka_client
+    kafka_consumer_check.event_platform_event = mock.Mock()
+
+    kafka_consumer_check._send_cluster_monitoring_heartbeat(total_contexts=0, cluster_id='test-cluster-id')
+
+    calls = kafka_consumer_check.event_platform_event.call_args_list
+    hb_events = [json.loads(c[0][0]) for c in calls if c[0][1] == 'data-streams-message']
+    hb_events = [e for e in hb_events if e.get('config_type') == 'heartbeat']
+    assert len(hb_events) == 1
+    assert hb_events[0]['brokers'] == []