Add out_of_sync_broker_id tag to partition metrics (DataDog#23428)

* Add out_of_sync_broker_id tag to Kafka partition metrics

When `enable_cluster_monitoring` is true, emit an `out_of_sync_broker_id:<id>`
tag on the kafka.partition.* metrics for each replica that is assigned to
the partition but not present in the ISR. Fully-replicated partitions are
unchanged (no new tags emitted), preserving existing dashboards and alerts.

This lets operators query kafka.partition.under_replicated (or the other
partition metrics) grouped by out_of_sync_broker_id to attribute under-
replicated partitions to specific broker IDs.

* Update style and test coverage to address feedback

* Assert out_of_sync_broker_id tag on all shared partition gauges

Author: chnn

kafka_consumer/changelog.d/23428.added

@@ -0,0 +1 @@
+Add an `out_of_sync_broker_id` tag to the `kafka.partition.*` metrics (when `enable_cluster_monitoring` is true) identifying each assigned replica that is not in the partition's ISR. Use it to attribute under-replicated partitions to specific broker IDs.

kafka_consumer/datadog_checks/kafka_consumer/cluster_metadata.py

@@ -571,12 +571,17 @@ def _collect_topic_metadata(self, metadata, highwater_offsets):
                     for replica in replicas:
                         partition_broker_tags.append(f'replica_broker_id:{replica}')
 
+                    isr_set = set(isrs)
+                    out_of_sync_broker_ids = [broker_id for broker_id in replicas if broker_id not in isr_set]
+                    for broker_id in out_of_sync_broker_ids:
+                        partition_broker_tags.append(f'out_of_sync_broker_id:{broker_id}')
+
                     self.check.gauge('partition.replicas', len(replicas), tags=partition_broker_tags)
                     self.check.gauge('partition.isr', len(isrs), tags=partition_broker_tags)
 
                     self.check.gauge('partition.size', partition_size, tags=partition_broker_tags)
 
-                    is_under_replicated = len(isrs) < len(replicas)
+                    is_under_replicated = bool(out_of_sync_broker_ids)
                     self.check.gauge(
                         'partition.under_replicated',
                         1 if is_under_replicated else 0,

kafka_consumer/tests/test_cluster_metadata.py

@@ -1351,3 +1351,59 @@ def test_schema_registry_url_encodes_subject_names(check):
     collector.http.get.assert_called_with(
         'http://localhost:8081/subjects/google%2Fprotobuf%2Ftimestamp.proto/versions/latest'
     )
+
+
+@pytest.mark.parametrize(
+    "replicas, isrs, expected_oos, expected_under",
+    [
+        pytest.param([1, 2], [1, 2], [], 0, id="fully_in_sync"),
+        pytest.param([1, 2], [1], [2], 1, id="single_oos"),
+        pytest.param([1, 2, 3], [1], [2, 3], 1, id="multiple_oos"),
+        pytest.param([1, 2], [], [1, 2], 1, id="empty_isr"),
+        pytest.param([1], [1], [], 0, id="single_replica"),
+    ],
+)
+def test_partition_out_of_sync_broker_id_tag(
+    check, dd_run_check, aggregator, replicas, isrs, expected_oos, expected_under
+):
+    """Under-replicated partitions expose an ``out_of_sync_broker_id`` tag per replica missing from the ISR."""
+    instance = {
+        'kafka_connect_str': 'localhost:9092',
+        'enable_cluster_monitoring': True,
+        'tags': ['test_tag:test_value'],
+    }
+
+    kafka_consumer_check = check(instance)
+    mock_kafka_client = seed_mock_kafka_client()
+
+    topic_metadata = mock_kafka_client.kafka_client.list_topics.return_value.topics['test-topic']
+    topic_metadata.partitions[0].replicas = replicas
+    topic_metadata.partitions[0].isrs = isrs
+
+    kafka_consumer_check.client = mock_kafka_client
+    kafka_consumer_check.metadata_collector.client = mock_kafka_client
+    mock_schema_registry_methods(kafka_consumer_check.metadata_collector)
+
+    kafka_consumer_check.read_persistent_cache = mock.Mock(return_value=None)
+    kafka_consumer_check.write_persistent_cache = mock.Mock()
+    kafka_consumer_check.event_platform_event = mock.Mock()
+
+    dd_run_check(kafka_consumer_check)
+
+    expected_tags = [
+        'test_tag:test_value',
+        'kafka_cluster_id:test-cluster-id',
+        'topic:test-topic',
+        'partition:0',
+        'leader_broker_id:1',
+        *(f'replica_broker_id:{r}' for r in replicas),
+        *(f'out_of_sync_broker_id:{b}' for b in expected_oos),
+    ]
+    aggregator.assert_metric('kafka.partition.under_replicated', value=expected_under, tags=expected_tags)
+    for metric in (
+        'kafka.partition.replicas',
+        'kafka.partition.isr',
+        'kafka.partition.size',
+        'kafka.partition.offline',
+    ):
+        aggregator.assert_metric(metric, tags=expected_tags)