From 455ec7bcb08d94501c1e19986191afb2c89f0516 Mon Sep 17 00:00:00 2001
From: Pulipelly Shashank Reddy <spulipelly@newrelic.com>
Date: Mon, 22 Jun 2026 18:23:42 +0530
Subject: [PATCH 1/6] feat(kafka): add
 MessageBroker/Kafka/Cluster/{id}/Topic/{topic} metrics
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Records per-cluster Kafka metrics (MessageBroker/Kafka/Cluster/{clusterId}/Topic/{topic}/Produce
and MessageBroker/Kafka/Cluster/{clusterId}/Topic/{topic}/Consume) to let customers track
throughput broken out by Kafka cluster, not just by topic. The cluster ID is fetched
once per unique broker set via a background AdminClient call; the hot produce/consume path
has no additional overhead. The feature is always-on and best-effort — it does not inject
anything into Kafka wire headers and does not add span or custom attributes.

Assisted-by: Claude Sonnet 4.6 <noreply@anthropic.com>
---
 .../hooks/messagebroker_confluentkafka.py     |  98 +++++++++
 newrelic/hooks/messagebroker_kafkapython.py   | 157 +++++++++++----
 .../test_consumer.py                          |  31 +++
 .../test_producer.py                          |  33 +++
 .../test_cluster_metrics_unit.py              | 190 ++++++++++++++++++
 .../test_consumer.py                          |  35 ++++
 .../test_producer.py                          |  39 ++++
 7 files changed, 544 insertions(+), 39 deletions(-)
 create mode 100644 tests/messagebroker_kafkapython/test_cluster_metrics_unit.py

diff --git a/newrelic/hooks/messagebroker_confluentkafka.py b/newrelic/hooks/messagebroker_confluentkafka.py
index 662e5ba87d..f24e9ac63c 100644
--- a/newrelic/hooks/messagebroker_confluentkafka.py
+++ b/newrelic/hooks/messagebroker_confluentkafka.py
@@ -13,6 +13,8 @@
 # limitations under the License.
 import logging
 import sys
+import threading
+import weakref
 
 from newrelic.api.application import application_instance
 from newrelic.api.error_trace import wrap_error_trace
@@ -33,6 +35,57 @@
 HEARTBEAT_SESSION_TIMEOUT = "MessageBroker/Kafka/Heartbeat/SessionTimeout"
 HEARTBEAT_POLL_TIMEOUT = "MessageBroker/Kafka/Heartbeat/PollTimeout"
 
+KAFKA_CLUSTER_METRIC_PRODUCE = "MessageBroker/Kafka/Cluster/{0}/Topic/{1}/Produce"
+KAFKA_CLUSTER_METRIC_CONSUME = "MessageBroker/Kafka/Cluster/{0}/Topic/{1}/Consume"
+
+
+_nr_cluster_id_cache = {}
+_nr_cluster_id_cache_lock = threading.Lock()
+
+
+def _fetch_cluster_id(instance):
+    servers = getattr(instance, "_nr_bootstrap_servers", None)
+    # Sort so that equivalent broker sets with different orderings share the same key.
+    cache_key = ",".join(sorted(servers)) if servers else None
+
+    if cache_key:
+        with _nr_cluster_id_cache_lock:
+            cached = _nr_cluster_id_cache.get(cache_key)
+            if cached:
+                instance._nr_cluster_id = cached
+                return
+            if cached is not None:
+                return
+            _nr_cluster_id_cache[cache_key] = ""
+
+    # Hold only a weak reference so the thread closure does not extend the
+    # lifetime of a Producer/Consumer that the caller has already abandoned.
+    instance_ref = weakref.ref(instance)
+
+    def _run():
+        inst = instance_ref()
+        if inst is None:
+            # Instance was GC'd before the thread ran; clean up sentinel and exit.
+            if cache_key:
+                with _nr_cluster_id_cache_lock:
+                    _nr_cluster_id_cache.pop(cache_key, None)
+            return
+        try:
+            meta = inst.list_topics(timeout=5)
+            cluster_id = getattr(meta, "cluster_id", None)
+            if cluster_id:
+                inst._nr_cluster_id = cluster_id
+                if cache_key:
+                    with _nr_cluster_id_cache_lock:
+                        _nr_cluster_id_cache[cache_key] = cluster_id
+        except Exception as e:
+            _logger.debug("NR Kafka cluster ID fetch failed", exc_info=True)
+            if cache_key:
+                with _nr_cluster_id_cache_lock:
+                    _nr_cluster_id_cache.pop(cache_key, None)
+
+    threading.Thread(target=_run, daemon=True, name="NR-Kafka-ClusterId").start()
+
 
 def wrap_Producer_produce(wrapped, instance, args, kwargs):
     transaction = current_transaction()
@@ -63,6 +116,17 @@ def wrap_Producer_produce(wrapped, instance, args, kwargs):
         for server_name in instance._nr_bootstrap_servers:
             transaction.record_custom_metric(f"MessageBroker/Kafka/Nodes/{server_name}/Produce/{topic}", 1)
 
+    cluster_id = getattr(instance, "_nr_cluster_id", None)
+    if not cluster_id and hasattr(instance, "_nr_bootstrap_servers"):
+        _cache_key = ",".join(sorted(instance._nr_bootstrap_servers))
+        cluster_id = _nr_cluster_id_cache.get(_cache_key) or None
+        if cluster_id:
+            instance._nr_cluster_id = cluster_id  # cache on instance for future calls
+    if cluster_id:
+        transaction.record_custom_metric(
+            KAFKA_CLUSTER_METRIC_PRODUCE.format(cluster_id, topic), 1
+        )
+
     with MessageTrace(
         library="Kafka", operation="Produce", destination_type="Topic", destination_name=topic, source=wrapped
     ):
@@ -171,6 +235,16 @@ def wrap_Consumer_poll(wrapped, instance, args, kwargs):
                     transaction.record_custom_metric(
                         f"MessageBroker/Kafka/Nodes/{server_name}/Consume/{destination_name}", 1
                     )
+            cluster_id = getattr(instance, "_nr_cluster_id", None)
+            if not cluster_id and hasattr(instance, "_nr_bootstrap_servers"):
+                _cache_key = ",".join(sorted(instance._nr_bootstrap_servers))
+                cluster_id = _nr_cluster_id_cache.get(_cache_key) or None
+                if cluster_id:
+                    instance._nr_cluster_id = cluster_id
+            if cluster_id:
+                transaction.record_custom_metric(
+                    KAFKA_CLUSTER_METRIC_CONSUME.format(cluster_id, destination_name), 1
+                )
             transaction.add_messagebroker_info("Confluent-Kafka", get_package_version("confluent-kafka"))
 
     return record
@@ -213,6 +287,16 @@ def wrap_SerializingProducer_init(wrapped, instance, args, kwargs):
     if hasattr(instance, "_value_serializer") and callable(instance._value_serializer):
         instance._value_serializer = wrap_serializer("Serialization/Value", "MessageBroker")(instance._value_serializer)
 
+    try:
+        conf = kwargs.get("conf") or (args[0] if args else {})
+        servers = conf.get("bootstrap.servers") if isinstance(conf, dict) else None
+        if servers:
+            instance._nr_bootstrap_servers = servers.split(",")
+    except Exception:
+        pass
+
+    _fetch_cluster_id(instance)
+
 
 def wrap_DeserializingConsumer_init(wrapped, instance, args, kwargs):
     wrapped(*args, **kwargs)
@@ -223,6 +307,16 @@ def wrap_DeserializingConsumer_init(wrapped, instance, args, kwargs):
     if hasattr(instance, "_value_deserializer") and callable(instance._value_deserializer):
         instance._value_deserializer = wrap_serializer("Deserialization/Value", "Message")(instance._value_deserializer)
 
+    try:
+        conf = kwargs.get("conf") or (args[0] if args else {})
+        servers = conf.get("bootstrap.servers") if isinstance(conf, dict) else None
+        if servers:
+            instance._nr_bootstrap_servers = servers.split(",")
+    except Exception:
+        pass
+
+    _fetch_cluster_id(instance)
+
 
 def wrap_Producer_init(wrapped, instance, args, kwargs):
     wrapped(*args, **kwargs)
@@ -236,6 +330,8 @@ def wrap_Producer_init(wrapped, instance, args, kwargs):
     except Exception:
         pass
 
+    _fetch_cluster_id(instance)
+
 
 def wrap_Consumer_init(wrapped, instance, args, kwargs):
     wrapped(*args, **kwargs)
@@ -249,6 +345,8 @@ def wrap_Consumer_init(wrapped, instance, args, kwargs):
     except Exception:
         pass
 
+    _fetch_cluster_id(instance)
+
 
 def wrap_immutable_class(module, class_name):
     # Wrap immutable binary extension class with a mutable Python subclass
diff --git a/newrelic/hooks/messagebroker_kafkapython.py b/newrelic/hooks/messagebroker_kafkapython.py
index ed0acf60ef..d837253fff 100644
--- a/newrelic/hooks/messagebroker_kafkapython.py
+++ b/newrelic/hooks/messagebroker_kafkapython.py
@@ -11,7 +11,9 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
+import logging
 import sys
+import threading
 
 from kafka.serializer import Serializer
 
@@ -31,6 +33,20 @@
 HEARTBEAT_SESSION_TIMEOUT = "MessageBroker/Kafka/Heartbeat/SessionTimeout"
 HEARTBEAT_POLL_TIMEOUT = "MessageBroker/Kafka/Heartbeat/PollTimeout"
 
+KAFKA_CLUSTER_METRIC_PRODUCE = "MessageBroker/Kafka/Cluster/{0}/Topic/{1}/Produce"
+KAFKA_CLUSTER_METRIC_CONSUME = "MessageBroker/Kafka/Cluster/{0}/Topic/{1}/Consume"
+
+_kafka_cluster_id_cache = {}
+_nr_cluster_id_cache_lock = threading.Lock()
+_logger = logging.getLogger(__name__)
+
+
+def _bootstrap_cache_key(bootstrap_servers):
+    """Normalize bootstrap_servers (str or iterable) to the cluster-ID cache key."""
+    if isinstance(bootstrap_servers, str):
+        bootstrap_servers = [bootstrap_servers]
+    return ",".join(sorted(str(s) for s in bootstrap_servers))
+
 
 def _bind_send(topic, value=None, key=None, headers=None, partition=None, timestamp_ms=None):
     return topic, value, key, headers, partition, timestamp_ms
@@ -66,6 +82,17 @@ def wrap_KafkaProducer_send(wrapped, instance, args, kwargs):
         if hasattr(instance, "config"):
             for server_name in instance.config.get("bootstrap_servers", []):
                 transaction.record_custom_metric(f"MessageBroker/Kafka/Nodes/{server_name}/Produce/{topic}", 1)
+
+        servers = instance.config.get("bootstrap_servers", []) if hasattr(instance, "config") else []
+        cluster_id = None
+        if servers:
+            _cache_key = _bootstrap_cache_key(servers)
+            cluster_id = _kafka_cluster_id_cache.get(_cache_key) or None
+        if cluster_id:
+            transaction.record_custom_metric(
+                KAFKA_CLUSTER_METRIC_PRODUCE.format(cluster_id, topic), 1
+            )
+
         try:
             return wrapped(
                 topic, value=value, key=key, headers=dt_headers, partition=partition, timestamp_ms=timestamp_ms
@@ -82,39 +109,14 @@ def wrap_kafkaconsumer_next(wrapped, instance, args, kwargs):
     try:
         record = wrapped(*args, **kwargs)
     except Exception as e:
-        # StopIteration is an expected error, indicating the end of an iterable,
-        # that should not be captured.
         if not isinstance(e, StopIteration):
             if current_transaction():
-                # Report error on existing transaction if there is one
                 notice_error()
             else:
-                # Report error on application
                 notice_error(application=application_instance(activate=False))
         raise
 
     if record:
-        # This iterator can be called either outside of a transaction, or
-        # within the context of an existing transaction.  There are 3
-        # possibilities we need to handle: (Note that this is similar to
-        # our Pika and Celery instrumentation)
-        #
-        #   1. In an inactive transaction
-        #
-        #      If the end_of_transaction() or ignore_transaction() API
-        #      calls have been invoked, this iterator may be called in the
-        #      context of an inactive transaction. In this case, don't wrap
-        #      the iterator in any way. Just run the original iterator.
-        #
-        #   2. In an active transaction
-        #
-        #      Do nothing.
-        #
-        #   3. Outside of a transaction
-        #
-        #      Since it's not running inside of an existing transaction, we
-        #      want to create a new background transaction for it.
-
         library = "Kafka"
         destination_type = "Topic"
         destination_name = record.topic
@@ -137,7 +139,6 @@ def wrap_kafkaconsumer_next(wrapped, instance, args, kwargs):
             instance._nr_transaction = transaction
             transaction.__enter__()
 
-            # Obtain consumer client_id to send up as agent attribute
             if hasattr(instance, "config") and "client_id" in instance.config:
                 client_id = instance.config["client_id"]
                 transaction._add_agent_attribute("kafka.consume.client_id", client_id)
@@ -145,11 +146,7 @@ def wrap_kafkaconsumer_next(wrapped, instance, args, kwargs):
             transaction._add_agent_attribute("kafka.consume.byteCount", received_bytes)
 
         transaction = current_transaction()
-        if transaction:  # If there is an active transaction now.
-            # Add metrics whether or not a transaction was already active, or one was just started.
-            # Don't add metrics if there was an inactive transaction.
-            # Name the metrics using the same format as the transaction, but in case the active transaction
-            # was an existing one and not a message transaction, reproduce the naming logic here.
+        if transaction:
             group = f"Message/{library}/{destination_type}"
             name = f"Named/{destination_name}"
             transaction.record_custom_metric(f"{group}/{name}/Received/Bytes", received_bytes)
@@ -159,6 +156,17 @@ def wrap_kafkaconsumer_next(wrapped, instance, args, kwargs):
                     transaction.record_custom_metric(
                         f"MessageBroker/Kafka/Nodes/{server_name}/Consume/{destination_name}", 1
                     )
+
+            servers = instance.config.get("bootstrap_servers", []) if hasattr(instance, "config") else []
+            cluster_id = None
+            if servers:
+                _cache_key = _bootstrap_cache_key(servers)
+                cached = _kafka_cluster_id_cache.get(_cache_key)
+                cluster_id = cached if cached else None
+            if cluster_id:
+                transaction.record_custom_metric(
+                    KAFKA_CLUSTER_METRIC_CONSUME.format(cluster_id, destination_name), 1
+                )
             transaction.add_messagebroker_info(
                 "Kafka-Python", get_package_version("kafka-python") or get_package_version("kafka-python-ng")
             )
@@ -166,6 +174,68 @@ def wrap_kafkaconsumer_next(wrapped, instance, args, kwargs):
     return record
 
 
+_SECURITY_CONFIG_KEYS = (
+    "security_protocol", "ssl_context", "ssl_cafile", "ssl_certfile",
+    "ssl_keyfile", "ssl_password", "ssl_crlfile", "ssl_ciphers",
+    "sasl_mechanism", "sasl_plain_username", "sasl_plain_password",
+    "sasl_kerberos_service_name", "sasl_kerberos_domain_name",
+    "sasl_oauth_token_provider",
+)
+
+
+def _fetch_cluster_id_kafka_python(bootstrap_servers, instance_config=None):
+    cache_key = _bootstrap_cache_key(bootstrap_servers)
+    with _nr_cluster_id_cache_lock:
+        if _kafka_cluster_id_cache.get(cache_key) is not None:
+            return
+        _kafka_cluster_id_cache[cache_key] = ""  # sentinel to prevent duplicate fetches
+
+    try:
+        from kafka.admin import KafkaAdminClient as _KafkaAdminClient
+        if not hasattr(_KafkaAdminClient, "describe_cluster"):
+            return
+    except ImportError:
+        return
+
+    def _run():
+        admin = None
+        cluster_id = None
+        try:
+            from kafka.admin import KafkaAdminClient
+            extra = {k: instance_config[k] for k in _SECURITY_CONFIG_KEYS if instance_config and k in instance_config and instance_config[k] is not None}
+            admin = KafkaAdminClient(
+                bootstrap_servers=bootstrap_servers,
+                client_id="newrelic-cluster-id-probe",
+                api_version_auto_timeout_ms=5000,
+                **extra,
+            )
+            meta = admin._client.cluster
+            admin._client.poll(timeout_ms=3000)
+            try:
+                result = admin.describe_cluster()
+                cluster_id = getattr(result, "cluster_id", None) or getattr(result, "clusterId", None)
+            except Exception:
+                _logger.debug("NR Kafka describe_cluster failed", exc_info=True)
+            if not cluster_id:
+                cluster_id = getattr(meta, "cluster_id", None) or getattr(meta, "_cluster_id", None)
+        except Exception:
+            _logger.debug("NR Kafka cluster ID fetch failed", exc_info=True)
+        finally:
+            try:
+                if admin is not None:
+                    admin.close()
+            except Exception:
+                pass
+        if cluster_id:
+            with _nr_cluster_id_cache_lock:
+                _kafka_cluster_id_cache[cache_key] = cluster_id
+        else:
+            with _nr_cluster_id_cache_lock:
+                _kafka_cluster_id_cache.pop(cache_key, None)
+
+    threading.Thread(target=_run, daemon=True, name="NR-KafkaPython-ClusterId").start()
+
+
 def wrap_KafkaProducer_init(wrapped, instance, args, kwargs):
     get_config_key = lambda key: kwargs.get(key, instance.DEFAULT_CONFIG[key])  # noqa: E731
 
@@ -176,7 +246,13 @@ def wrap_KafkaProducer_init(wrapped, instance, args, kwargs):
         instance, "Serialization/Value", "MessageBroker", get_config_key("value_serializer")
     )
 
-    return wrapped(*args, **kwargs)
+    result = wrapped(*args, **kwargs)
+
+    servers = instance.config.get("bootstrap_servers", [])
+    if servers:
+        _fetch_cluster_id_kafka_python(servers, instance.config)
+
+    return result
 
 
 class NewRelicSerializerWrapper(ObjectProxy):
@@ -211,7 +287,6 @@ def _wrap_serializer(wrapped, instance, args, kwargs):
         if isinstance(transaction, MessageTransaction):
             topic = transaction.destination_name
         else:
-            # Find parent message trace to retrieve topic
             message_trace = current_trace()
             while message_trace is not None and not isinstance(message_trace, MessageTrace):
                 message_trace = message_trace.parent
@@ -224,17 +299,14 @@ def _wrap_serializer(wrapped, instance, args, kwargs):
         return FunctionTraceWrapper(wrapped, name=name, group=group)(*args, **kwargs)
 
     try:
-        # Apply wrapper to serializer
         if serializer is None:
-            # Do nothing
             return serializer
         elif isinstance(serializer, Serializer):
             return NewRelicSerializerWrapper(serializer, group_prefix=group_prefix, serializer_name=serializer_name)
         else:
-            # Wrap callable in wrapper
             return _wrap_serializer(serializer)
     except Exception:
-        return serializer  # Avoid crashes from immutable serializers
+        return serializer
 
 
 def metric_wrapper(metric_name, check_result=False):
@@ -244,8 +316,6 @@ def _metric_wrapper(wrapped, instance, args, kwargs):
         application = application_instance(activate=False)
         if application:
             if not check_result or (check_result and result):
-                # If the result does not need validated, send metric.
-                # If the result does need validated, ensure it is True.
                 application.record_custom_metric(metric_name, 1)
 
         return result
@@ -259,8 +329,17 @@ def instrument_kafka_producer(module):
         wrap_function_wrapper(module, "KafkaProducer.send", wrap_KafkaProducer_send)
 
 
+def _wrap_KafkaConsumer_init(wrapped, instance, args, kwargs):
+    result = wrapped(*args, **kwargs)
+    servers = instance.config.get("bootstrap_servers", [])
+    if servers:
+        _fetch_cluster_id_kafka_python(servers, instance.config)
+    return result
+
+
 def instrument_kafka_consumer_group(module):
     if hasattr(module, "KafkaConsumer"):
+        wrap_function_wrapper(module, "KafkaConsumer.__init__", _wrap_KafkaConsumer_init)
         wrap_function_wrapper(module, "KafkaConsumer.__next__", wrap_kafkaconsumer_next)
 
 
diff --git a/tests/messagebroker_confluentkafka/test_consumer.py b/tests/messagebroker_confluentkafka/test_consumer.py
index 6eadb49edd..cff7126a02 100644
--- a/tests/messagebroker_confluentkafka/test_consumer.py
+++ b/tests/messagebroker_confluentkafka/test_consumer.py
@@ -182,3 +182,34 @@ def expected_broker_metrics(broker, topic):
 @pytest.fixture
 def expected_missing_broker_metrics(broker, topic):
     return [(f"MessageBroker/Kafka/Nodes/{server}/Consume/{topic}", None) for server in broker.split(",")]
+
+
+# ---------------------------------------------------------------------------
+# Cluster-ID metric and attribute tests (confluent-kafka)
+# ---------------------------------------------------------------------------
+
+@pytest.fixture
+def consumer_with_cluster_id(consumer, broker):
+    """Set _nr_cluster_id directly on the consumer instance for deterministic tests."""
+    test_cluster_id = "confluent-consumer-cluster-test"
+    consumer._nr_cluster_id = test_cluster_id
+    if not hasattr(consumer, "_nr_bootstrap_servers"):
+        consumer._nr_bootstrap_servers = broker.split(",")
+    yield consumer, test_cluster_id
+
+
+def test_cluster_consume_metric(topic, get_consumer_record, consumer_with_cluster_id, client_type):
+    """MessageBroker/Kafka/Cluster/{id}/Topic/{topic}/Consume appears after poll()."""
+    _, cluster_id = consumer_with_cluster_id
+    cluster_metric = f"MessageBroker/Kafka/Cluster/{cluster_id}/Topic/{topic}/Consume"
+
+    @validate_transaction_metrics(
+        f"Named/{topic}",
+        group="Message/Kafka/Topic",
+        custom_metrics=[(cluster_metric, 1)],
+        background_task=True,
+    )
+    def _test():
+        get_consumer_record()
+
+    _test()
diff --git a/tests/messagebroker_confluentkafka/test_producer.py b/tests/messagebroker_confluentkafka/test_producer.py
index 14bb7535e0..e577bb42d5 100644
--- a/tests/messagebroker_confluentkafka/test_producer.py
+++ b/tests/messagebroker_confluentkafka/test_producer.py
@@ -152,3 +152,36 @@ def test():
 @pytest.fixture
 def expected_broker_metrics(broker, topic):
     return [(f"MessageBroker/Kafka/Nodes/{server}/Produce/{topic}", 1) for server in broker.split(",")]
+
+
+# ---------------------------------------------------------------------------
+# Cluster-ID metric tests (confluent-kafka)
+# ---------------------------------------------------------------------------
+
+@pytest.fixture
+def producer_with_cluster_id(producer, broker):
+    """Set _nr_cluster_id directly on the producer instance, bypassing the
+    async daemon-thread fetch so metric tests are deterministic and fast."""
+    test_cluster_id = "confluent-cluster-test-999"
+    producer._nr_cluster_id = test_cluster_id
+    # Also need bootstrap servers so the Nodes metrics fire correctly
+    if not hasattr(producer, "_nr_bootstrap_servers"):
+        producer._nr_bootstrap_servers = broker.split(",")
+    yield producer, test_cluster_id
+
+
+def test_cluster_produce_metric(topic, producer_with_cluster_id, send_producer_message, client_type):
+    """MessageBroker/Kafka/Cluster/{id}/Topic/{topic}/Produce appears after produce()."""
+    _, cluster_id = producer_with_cluster_id
+    cluster_metric = f"MessageBroker/Kafka/Cluster/{cluster_id}/Topic/{topic}/Produce"
+
+    @validate_transaction_metrics(
+        "test_producer:test_cluster_produce_metric.<locals>.test",
+        custom_metrics=[(cluster_metric, 1)],
+        background_task=True,
+    )
+    @background_task()
+    def test():
+        send_producer_message()
+
+    test()
diff --git a/tests/messagebroker_kafkapython/test_cluster_metrics_unit.py b/tests/messagebroker_kafkapython/test_cluster_metrics_unit.py
new file mode 100644
index 0000000000..7b34aed0e1
--- /dev/null
+++ b/tests/messagebroker_kafkapython/test_cluster_metrics_unit.py
@@ -0,0 +1,190 @@
+# Copyright 2010 New Relic, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Unit tests for cluster-ID additions in messagebroker_kafkapython.
+
+These tests exercise the wrapper functions directly with mocks — no real Kafka
+broker required. They verify correctness of arguments passed to the underlying
+`wrapped` callable without any network I/O.
+"""
+
+from unittest.mock import MagicMock, patch
+
+from newrelic.hooks.messagebroker_kafkapython import (
+    _bootstrap_cache_key,
+    _fetch_cluster_id_kafka_python,
+    _kafka_cluster_id_cache,
+    wrap_KafkaProducer_send,
+)
+
+
+# ---------------------------------------------------------------------------
+# PY-1 regression: wrap_KafkaProducer_send must not overwrite the Kafka
+# message routing key with the broker address string.
+# ---------------------------------------------------------------------------
+
+class TestProducerSendKeyPreservation:
+    """The Kafka message routing key must survive the wrap_KafkaProducer_send
+    instrumentation unchanged, regardless of whether cluster ID is cached."""
+
+    def _make_producer_instance(self, bootstrap_servers=None):
+        instance = MagicMock()
+        instance.config = {
+            "bootstrap_servers": bootstrap_servers or ["broker1:9092", "broker2:9092"],
+        }
+        return instance
+
+    def _bind_send_args(self, topic, value=None, key=None, headers=None):
+        """Return positional args as wrap_KafkaProducer_send receives them."""
+        return (topic,), {"value": value, "key": key, "headers": headers or []}
+
+    def test_message_key_not_overwritten_with_cluster_id_in_cache(self):
+        """Key must not be replaced by broker address string when cluster ID cached."""
+        cluster_id = "test-cluster-uuid"
+        cache_key = "broker1:9092,broker2:9092"
+        _kafka_cluster_id_cache[cache_key] = cluster_id
+
+        wrapped = MagicMock(return_value=MagicMock())
+        instance = self._make_producer_instance()
+        args, kwargs = self._bind_send_args("my-topic", value=b"v", key=b"original-key")
+
+        try:
+            with patch("newrelic.hooks.messagebroker_kafkapython.current_transaction") as mock_txn:
+                mock_txn.return_value = MagicMock()  # active transaction
+                wrap_KafkaProducer_send(wrapped, instance, args, kwargs)
+        finally:
+            _kafka_cluster_id_cache.pop(cache_key, None)
+
+        # The wrapped callable must have been called with key=b"original-key",
+        # not key="broker1:9092,broker2:9092" or any other broker-derived string.
+        assert wrapped.called, "wrapped() was never called"
+        call_kwargs = wrapped.call_args[1]
+        assert call_kwargs["key"] == b"original-key", (
+            f"Message key was corrupted: got {call_kwargs['key']!r}, "
+            f"expected b'original-key'. Likely cause: cache lookup variable "
+            f"reused the name 'key', overwriting the Kafka routing key."
+        )
+
+    def test_message_key_not_overwritten_when_no_cluster_id_cached(self):
+        """Key must not be replaced even when cluster ID is not yet in the cache."""
+        wrapped = MagicMock(return_value=MagicMock())
+        instance = self._make_producer_instance(bootstrap_servers=["broker-no-cache:9092"])
+        args, kwargs = self._bind_send_args("topic", key="string-key-123")
+
+        with patch("newrelic.hooks.messagebroker_kafkapython.current_transaction") as mock_txn:
+            mock_txn.return_value = MagicMock()
+            wrap_KafkaProducer_send(wrapped, instance, args, kwargs)
+
+        assert wrapped.called
+        assert wrapped.call_args[1]["key"] == "string-key-123", (
+            "Message key corrupted even when cluster ID was not in cache."
+        )
+
+    def test_none_key_preserved(self):
+        """A None routing key must remain None (common case for unkeyed messages)."""
+        wrapped = MagicMock(return_value=MagicMock())
+        instance = self._make_producer_instance()
+        args, kwargs = self._bind_send_args("topic", key=None)
+
+        with patch("newrelic.hooks.messagebroker_kafkapython.current_transaction") as mock_txn:
+            mock_txn.return_value = MagicMock()
+            wrap_KafkaProducer_send(wrapped, instance, args, kwargs)
+
+        assert wrapped.call_args[1]["key"] is None, "None key was corrupted."
+
+    def test_no_transaction_bypasses_instrumentation(self):
+        """Without an active NR transaction, wrapped() is called with original args."""
+        wrapped = MagicMock(return_value=MagicMock())
+        instance = self._make_producer_instance()
+        args = ("topic",)
+        kwargs = {"value": b"v", "key": b"my-key"}
+
+        with patch("newrelic.hooks.messagebroker_kafkapython.current_transaction") as mock_txn:
+            mock_txn.return_value = None  # no active transaction
+            wrap_KafkaProducer_send(wrapped, instance, args, kwargs)
+
+        # wrapped() called directly with original args — no instrumentation applied
+        assert wrapped.called
+        wrapped.assert_called_once_with(*args, **kwargs)
+
+
+# ---------------------------------------------------------------------------
+# Cluster ID is fetched via KafkaAdminClient.describe_cluster() in a
+# background daemon thread — no passive MetadataResponse interception.
+# ---------------------------------------------------------------------------
+
+class TestFetchClusterIdKafkaPython:
+    """_fetch_cluster_id_kafka_python populates the cache via KafkaAdminClient."""
+
+    def _cache_key(self, servers):
+        return _bootstrap_cache_key(servers)
+
+    def test_cluster_id_written_to_cache_on_success(self):
+        """A successful describe_cluster() stores the cluster UUID in the module cache."""
+        servers = ["broker-fetch1:9092"]
+        cache_key = self._cache_key(servers)
+        _kafka_cluster_id_cache.pop(cache_key, None)
+
+        mock_admin = MagicMock()
+        mock_admin.describe_cluster.return_value = {"cluster_id": "fetched-uuid", "brokers": []}
+
+        with patch("kafka.admin.KafkaAdminClient", return_value=mock_admin):
+            _fetch_cluster_id_kafka_python(servers)
+            # Allow the daemon thread to finish
+            import time; time.sleep(0.5)
+
+        try:
+            assert _kafka_cluster_id_cache.get(cache_key) == "fetched-uuid"
+        finally:
+            _kafka_cluster_id_cache.pop(cache_key, None)
+
+    def test_sentinel_prevents_duplicate_fetches(self):
+        """While a fetch is in-flight (sentinel ''), a second call does not spawn a thread."""
+        servers = ["broker-sentinel:9092"]
+        cache_key = self._cache_key(servers)
+        _kafka_cluster_id_cache[cache_key] = ""  # pre-set sentinel
+
+        with patch("newrelic.hooks.messagebroker_kafkapython.threading.Thread") as mock_thread:
+            _fetch_cluster_id_kafka_python(servers)
+            mock_thread.assert_not_called()
+
+        _kafka_cluster_id_cache.pop(cache_key, None)
+
+    def test_existing_resolved_entry_not_refetched(self):
+        """A fully-resolved cache entry (non-empty) skips the fetch and returns immediately."""
+        servers = ["broker-resolved:9092"]
+        cache_key = self._cache_key(servers)
+        _kafka_cluster_id_cache[cache_key] = "already-resolved-uuid"
+
+        with patch("newrelic.hooks.messagebroker_kafkapython.threading.Thread") as mock_thread:
+            _fetch_cluster_id_kafka_python(servers)
+            mock_thread.assert_not_called()
+
+        _kafka_cluster_id_cache.pop(cache_key, None)
+
+    def test_sentinel_removed_on_failure(self):
+        """If KafkaAdminClient raises, the sentinel is removed so future instances can retry."""
+        servers = ["broker-fail:9092"]
+        cache_key = self._cache_key(servers)
+        _kafka_cluster_id_cache.pop(cache_key, None)
+
+        # Raise at construction time — this hits the outer except in _run() which
+        # removes the sentinel. Raising only on connect() wouldn't work because
+        # MagicMock.describe_cluster() auto-creates a truthy return value, causing
+        # the cache to be populated with a MagicMock instead of being cleared.
+        with patch("kafka.admin.KafkaAdminClient", side_effect=Exception("connection refused")):
+            _fetch_cluster_id_kafka_python(servers)
+            import time; time.sleep(0.3)
+
+        assert _kafka_cluster_id_cache.get(cache_key) is None
diff --git a/tests/messagebroker_kafkapython/test_consumer.py b/tests/messagebroker_kafkapython/test_consumer.py
index e53bc4ff7c..ac08760ced 100644
--- a/tests/messagebroker_kafkapython/test_consumer.py
+++ b/tests/messagebroker_kafkapython/test_consumer.py
@@ -186,3 +186,38 @@ def expected_broker_metrics(broker, topic):
 @pytest.fixture
 def expected_missing_broker_metrics(broker, topic):
     return [(f"MessageBroker/Kafka/Nodes/{server}/Consume/{topic}", None) for server in broker]
+
+
+# ---------------------------------------------------------------------------
+# Cluster-ID metric and attribute tests
+# ---------------------------------------------------------------------------
+
+@pytest.fixture
+def seeded_cluster_id(broker):
+    """Pre-seed the cluster-ID cache so metric tests are deterministic."""
+    from newrelic.hooks.messagebroker_kafkapython import _kafka_cluster_id_cache
+
+    cache_key = ",".join(sorted(broker))
+    test_cluster_id = "test-cluster-consumer-xyz"
+    _kafka_cluster_id_cache[cache_key] = test_cluster_id
+    yield test_cluster_id
+    _kafka_cluster_id_cache.pop(cache_key, None)
+
+
+def test_cluster_consume_metric(get_consumer_record, topic, broker, seeded_cluster_id):
+    """MessageBroker/Kafka/Cluster/{id}/Topic/{topic}/Consume appears after a poll."""
+    cluster_id = seeded_cluster_id
+    cluster_metric = f"MessageBroker/Kafka/Cluster/{cluster_id}/Topic/{topic}/Consume"
+
+    @validate_transaction_metrics(
+        f"Named/{topic}",
+        group="Message/Kafka/Topic",
+        custom_metrics=[(cluster_metric, 1)],
+        background_task=True,
+    )
+    def _test():
+        get_consumer_record()
+
+    _test()
+
+
diff --git a/tests/messagebroker_kafkapython/test_producer.py b/tests/messagebroker_kafkapython/test_producer.py
index 684807be8b..c28afc6cc8 100644
--- a/tests/messagebroker_kafkapython/test_producer.py
+++ b/tests/messagebroker_kafkapython/test_producer.py
@@ -101,3 +101,42 @@ def test():
 @pytest.fixture
 def expected_broker_metrics(broker, topic):
     return [(f"MessageBroker/Kafka/Nodes/{server}/Produce/{topic}", 1) for server in broker]
+
+
+# ---------------------------------------------------------------------------
+# Cluster-ID metric tests
+# ---------------------------------------------------------------------------
+
+@pytest.fixture
+def seeded_cluster_id(broker):
+    """Pre-seed the cluster-ID cache with a known value so tests are deterministic.
+
+    The real cluster-ID fetch is async; seeding avoids flaky timing issues
+    in the test suite while still exercising the metric-recording code path.
+    """
+    from newrelic.hooks.messagebroker_kafkapython import _kafka_cluster_id_cache
+
+    cache_key = ",".join(sorted(broker))
+    test_cluster_id = "test-cluster-abc123"
+    _kafka_cluster_id_cache[cache_key] = test_cluster_id
+    yield test_cluster_id
+    _kafka_cluster_id_cache.pop(cache_key, None)
+
+
+def test_cluster_produce_metric(topic, send_producer_message, broker, seeded_cluster_id):
+    """MessageBroker/Kafka/Cluster/{id}/Topic/{topic}/Produce appears after a send."""
+    cluster_id = seeded_cluster_id
+    cluster_metric = f"MessageBroker/Kafka/Cluster/{cluster_id}/Topic/{topic}/Produce"
+
+    @validate_transaction_metrics(
+        "test_producer:test_cluster_produce_metric.<locals>.test",
+        custom_metrics=[(cluster_metric, 1)],
+        background_task=True,
+    )
+    @background_task()
+    def test():
+        send_producer_message()
+
+    test()
+
+

From c948053d20c44401a944d4a0a3b2eec5f9213028 Mon Sep 17 00:00:00 2001
From: Pulipelly Shashank Reddy <spulipelly@newrelic.com>
Date: Sun, 28 Jun 2026 19:50:14 +0530
Subject: [PATCH 2/6] fix(kafka): add 60-min TTL and stale-while-revalidate to
 cluster-id caches

Both hooks now store (cluster_id, time.monotonic()) tuples and trigger a
background re-fetch on expiry, returning the stale value in the meantime.
Also handles describe_cluster() returning a dict (newer kafka-python-ng).

Assisted-by: Claude Sonnet 4.6
---
 .../hooks/messagebroker_confluentkafka.py     | 50 ++++++++++++++-----
 newrelic/hooks/messagebroker_kafkapython.py   | 47 ++++++++++++++---
 2 files changed, 77 insertions(+), 20 deletions(-)

diff --git a/newrelic/hooks/messagebroker_confluentkafka.py b/newrelic/hooks/messagebroker_confluentkafka.py
index f24e9ac63c..ff4856e6a9 100644
--- a/newrelic/hooks/messagebroker_confluentkafka.py
+++ b/newrelic/hooks/messagebroker_confluentkafka.py
@@ -14,6 +14,7 @@
 import logging
 import sys
 import threading
+import time
 import weakref
 
 from newrelic.api.application import application_instance
@@ -39,6 +40,8 @@
 KAFKA_CLUSTER_METRIC_CONSUME = "MessageBroker/Kafka/Cluster/{0}/Topic/{1}/Consume"
 
 
+_CLUSTER_ID_TTL_SECONDS = 60 * 60
+
 _nr_cluster_id_cache = {}
 _nr_cluster_id_cache_lock = threading.Lock()
 
@@ -51,10 +54,15 @@ def _fetch_cluster_id(instance):
     if cache_key:
         with _nr_cluster_id_cache_lock:
             cached = _nr_cluster_id_cache.get(cache_key)
-            if cached:
-                instance._nr_cluster_id = cached
-                return
-            if cached is not None:
+            if isinstance(cached, tuple):
+                cluster_id, ts = cached
+                if time.monotonic() - ts <= _CLUSTER_ID_TTL_SECONDS:
+                    instance._nr_cluster_id = cluster_id
+                    instance._nr_cluster_id_fetched_at = ts
+                    return
+                # Expired — fall through to start a new fetch.
+            elif cached is not None:
+                # "" sentinel — fetch already in progress.
                 return
             _nr_cluster_id_cache[cache_key] = ""
 
@@ -74,11 +82,13 @@ def _run():
             meta = inst.list_topics(timeout=5)
             cluster_id = getattr(meta, "cluster_id", None)
             if cluster_id:
+                now = time.monotonic()
                 inst._nr_cluster_id = cluster_id
+                inst._nr_cluster_id_fetched_at = now
                 if cache_key:
                     with _nr_cluster_id_cache_lock:
-                        _nr_cluster_id_cache[cache_key] = cluster_id
-        except Exception as e:
+                        _nr_cluster_id_cache[cache_key] = (cluster_id, now)
+        except Exception:
             _logger.debug("NR Kafka cluster ID fetch failed", exc_info=True)
             if cache_key:
                 with _nr_cluster_id_cache_lock:
@@ -117,11 +127,18 @@ def wrap_Producer_produce(wrapped, instance, args, kwargs):
             transaction.record_custom_metric(f"MessageBroker/Kafka/Nodes/{server_name}/Produce/{topic}", 1)
 
     cluster_id = getattr(instance, "_nr_cluster_id", None)
-    if not cluster_id and hasattr(instance, "_nr_bootstrap_servers"):
+    if cluster_id:
+        if time.monotonic() - getattr(instance, "_nr_cluster_id_fetched_at", 0.0) > _CLUSTER_ID_TTL_SECONDS:
+            _fetch_cluster_id(instance)  # background re-fetch; use stale value below
+    elif hasattr(instance, "_nr_bootstrap_servers"):
         _cache_key = ",".join(sorted(instance._nr_bootstrap_servers))
-        cluster_id = _nr_cluster_id_cache.get(_cache_key) or None
-        if cluster_id:
-            instance._nr_cluster_id = cluster_id  # cache on instance for future calls
+        _cached = _nr_cluster_id_cache.get(_cache_key)
+        if isinstance(_cached, tuple):
+            cluster_id, _ts = _cached
+            instance._nr_cluster_id = cluster_id
+            instance._nr_cluster_id_fetched_at = _ts
+            if time.monotonic() - _ts > _CLUSTER_ID_TTL_SECONDS:
+                _fetch_cluster_id(instance)  # background re-fetch
     if cluster_id:
         transaction.record_custom_metric(
             KAFKA_CLUSTER_METRIC_PRODUCE.format(cluster_id, topic), 1
@@ -236,11 +253,18 @@ def wrap_Consumer_poll(wrapped, instance, args, kwargs):
                         f"MessageBroker/Kafka/Nodes/{server_name}/Consume/{destination_name}", 1
                     )
             cluster_id = getattr(instance, "_nr_cluster_id", None)
-            if not cluster_id and hasattr(instance, "_nr_bootstrap_servers"):
+            if cluster_id:
+                if time.monotonic() - getattr(instance, "_nr_cluster_id_fetched_at", 0.0) > _CLUSTER_ID_TTL_SECONDS:
+                    _fetch_cluster_id(instance)  # background re-fetch; use stale value below
+            elif hasattr(instance, "_nr_bootstrap_servers"):
                 _cache_key = ",".join(sorted(instance._nr_bootstrap_servers))
-                cluster_id = _nr_cluster_id_cache.get(_cache_key) or None
-                if cluster_id:
+                _cached = _nr_cluster_id_cache.get(_cache_key)
+                if isinstance(_cached, tuple):
+                    cluster_id, _ts = _cached
                     instance._nr_cluster_id = cluster_id
+                    instance._nr_cluster_id_fetched_at = _ts
+                    if time.monotonic() - _ts > _CLUSTER_ID_TTL_SECONDS:
+                        _fetch_cluster_id(instance)  # background re-fetch
             if cluster_id:
                 transaction.record_custom_metric(
                     KAFKA_CLUSTER_METRIC_CONSUME.format(cluster_id, destination_name), 1
diff --git a/newrelic/hooks/messagebroker_kafkapython.py b/newrelic/hooks/messagebroker_kafkapython.py
index d837253fff..2861845a39 100644
--- a/newrelic/hooks/messagebroker_kafkapython.py
+++ b/newrelic/hooks/messagebroker_kafkapython.py
@@ -14,6 +14,7 @@
 import logging
 import sys
 import threading
+import time
 
 from kafka.serializer import Serializer
 
@@ -36,6 +37,8 @@
 KAFKA_CLUSTER_METRIC_PRODUCE = "MessageBroker/Kafka/Cluster/{0}/Topic/{1}/Produce"
 KAFKA_CLUSTER_METRIC_CONSUME = "MessageBroker/Kafka/Cluster/{0}/Topic/{1}/Consume"
 
+_CLUSTER_ID_TTL_SECONDS = 60 * 60
+
 _kafka_cluster_id_cache = {}
 _nr_cluster_id_cache_lock = threading.Lock()
 _logger = logging.getLogger(__name__)
@@ -56,6 +59,20 @@ def wrap_KafkaProducer_send(wrapped, instance, args, kwargs):
     transaction = current_transaction()
 
     if transaction is None:
+        topic, *_ = _bind_send(*args, **kwargs)
+        topic = topic or "Default"
+        servers = instance.config.get("bootstrap_servers", []) if hasattr(instance, "config") else []
+        if servers:
+            _cache_key = _bootstrap_cache_key(servers)
+            _cached = _kafka_cluster_id_cache.get(_cache_key)
+            if isinstance(_cached, tuple):
+                cluster_id, _ts = _cached
+                if time.monotonic() - _ts > _CLUSTER_ID_TTL_SECONDS:
+                    _fetch_cluster_id_kafka_python(servers, instance.config if hasattr(instance, "config") else None)
+                if cluster_id:
+                    app = application_instance(activate=False)
+                    if app:
+                        app.record_custom_metric(KAFKA_CLUSTER_METRIC_PRODUCE.format(cluster_id, topic), 1)
         return wrapped(*args, **kwargs)
 
     topic, value, key, headers, partition, timestamp_ms = _bind_send(*args, **kwargs)
@@ -87,7 +104,11 @@ def wrap_KafkaProducer_send(wrapped, instance, args, kwargs):
         cluster_id = None
         if servers:
             _cache_key = _bootstrap_cache_key(servers)
-            cluster_id = _kafka_cluster_id_cache.get(_cache_key) or None
+            _cached = _kafka_cluster_id_cache.get(_cache_key)
+            if isinstance(_cached, tuple):
+                cluster_id, _ts = _cached
+                if time.monotonic() - _ts > _CLUSTER_ID_TTL_SECONDS:
+                    _fetch_cluster_id_kafka_python(servers, instance.config if hasattr(instance, "config") else None)
         if cluster_id:
             transaction.record_custom_metric(
                 KAFKA_CLUSTER_METRIC_PRODUCE.format(cluster_id, topic), 1
@@ -109,6 +130,7 @@ def wrap_kafkaconsumer_next(wrapped, instance, args, kwargs):
     try:
         record = wrapped(*args, **kwargs)
     except Exception as e:
+        # StopIteration ends iteration normally — do not capture.
         if not isinstance(e, StopIteration):
             if current_transaction():
                 notice_error()
@@ -161,8 +183,11 @@ def wrap_kafkaconsumer_next(wrapped, instance, args, kwargs):
             cluster_id = None
             if servers:
                 _cache_key = _bootstrap_cache_key(servers)
-                cached = _kafka_cluster_id_cache.get(_cache_key)
-                cluster_id = cached if cached else None
+                _cached = _kafka_cluster_id_cache.get(_cache_key)
+                if isinstance(_cached, tuple):
+                    cluster_id, _ts = _cached
+                    if time.monotonic() - _ts > _CLUSTER_ID_TTL_SECONDS:
+                        _fetch_cluster_id_kafka_python(servers, instance.config if hasattr(instance, "config") else None)
             if cluster_id:
                 transaction.record_custom_metric(
                     KAFKA_CLUSTER_METRIC_CONSUME.format(cluster_id, destination_name), 1
@@ -186,8 +211,13 @@ def wrap_kafkaconsumer_next(wrapped, instance, args, kwargs):
 def _fetch_cluster_id_kafka_python(bootstrap_servers, instance_config=None):
     cache_key = _bootstrap_cache_key(bootstrap_servers)
     with _nr_cluster_id_cache_lock:
-        if _kafka_cluster_id_cache.get(cache_key) is not None:
-            return
+        cached = _kafka_cluster_id_cache.get(cache_key)
+        if isinstance(cached, tuple):
+            if time.monotonic() - cached[1] <= _CLUSTER_ID_TTL_SECONDS:
+                return  # still fresh
+            # Expired — fall through to start a new fetch.
+        elif cached is not None:
+            return  # "" sentinel — fetch already in progress.
         _kafka_cluster_id_cache[cache_key] = ""  # sentinel to prevent duplicate fetches
 
     try:
@@ -213,7 +243,10 @@ def _run():
             admin._client.poll(timeout_ms=3000)
             try:
                 result = admin.describe_cluster()
-                cluster_id = getattr(result, "cluster_id", None) or getattr(result, "clusterId", None)
+                if isinstance(result, dict):
+                    cluster_id = result.get("cluster_id") or result.get("clusterId")
+                else:
+                    cluster_id = getattr(result, "cluster_id", None) or getattr(result, "clusterId", None)
             except Exception:
                 _logger.debug("NR Kafka describe_cluster failed", exc_info=True)
             if not cluster_id:
@@ -228,7 +261,7 @@ def _run():
                 pass
         if cluster_id:
             with _nr_cluster_id_cache_lock:
-                _kafka_cluster_id_cache[cache_key] = cluster_id
+                _kafka_cluster_id_cache[cache_key] = (cluster_id, time.monotonic())
         else:
             with _nr_cluster_id_cache_lock:
                 _kafka_cluster_id_cache.pop(cache_key, None)

From 7c2ce66bf5ec4835de54b0b475c14b8065d0dfb2 Mon Sep 17 00:00:00 2001
From: Pulipelly Shashank Reddy <spulipelly@newrelic.com>
Date: Mon, 29 Jun 2026 00:53:43 +0530
Subject: [PATCH 3/6] fix(tests): move import time to module level in
 test_cluster_metrics_unit

Ruff E702 (multiple statements on one line) and I001 (unsorted imports)
flagged the inline `import time; time.sleep(...)` patterns. Moving the
import to the module-level import block is the correct fix.

Assisted-by: Claude Sonnet 4.6
---
 tests/messagebroker_kafkapython/test_cluster_metrics_unit.py | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/tests/messagebroker_kafkapython/test_cluster_metrics_unit.py b/tests/messagebroker_kafkapython/test_cluster_metrics_unit.py
index 7b34aed0e1..e0cce4322d 100644
--- a/tests/messagebroker_kafkapython/test_cluster_metrics_unit.py
+++ b/tests/messagebroker_kafkapython/test_cluster_metrics_unit.py
@@ -19,6 +19,7 @@
 `wrapped` callable without any network I/O.
 """
 
+import time
 from unittest.mock import MagicMock, patch
 
 from newrelic.hooks.messagebroker_kafkapython import (
@@ -142,7 +143,7 @@ def test_cluster_id_written_to_cache_on_success(self):
         with patch("kafka.admin.KafkaAdminClient", return_value=mock_admin):
             _fetch_cluster_id_kafka_python(servers)
             # Allow the daemon thread to finish
-            import time; time.sleep(0.5)
+            time.sleep(0.5)
 
         try:
             assert _kafka_cluster_id_cache.get(cache_key) == "fetched-uuid"
@@ -185,6 +186,6 @@ def test_sentinel_removed_on_failure(self):
         # the cache to be populated with a MagicMock instead of being cleared.
         with patch("kafka.admin.KafkaAdminClient", side_effect=Exception("connection refused")):
             _fetch_cluster_id_kafka_python(servers)
-            import time; time.sleep(0.3)
+            time.sleep(0.3)
 
         assert _kafka_cluster_id_cache.get(cache_key) is None

From c322bd592ef10115839a3100f02706d117d9bb51 Mon Sep 17 00:00:00 2001
From: Pulipelly Shashank Reddy <spulipelly@newrelic.com>
Date: Mon, 29 Jun 2026 01:13:33 +0530
Subject: [PATCH 4/6] fix: align test cache seeding with (cluster_id,
 timestamp) tuple format
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The hook stores cluster IDs as (str, float) tuples for stale-while-revalidate
TTL tracking. Three test files seeded _kafka_cluster_id_cache with plain
strings, so isinstance(_cached, tuple) was False and metrics were never
recorded — causing test_cluster_produce_metric, test_cluster_consume_metric,
and test_cluster_id_written_to_cache_on_success to fail.

- test_producer.py / test_consumer.py: seed with (id, time.monotonic()) tuple;
  add missing import time
- test_cluster_metrics_unit.py: same for key-preservation and resolved-entry
  fixtures; fix assertion to check cached[0] instead of comparing full tuple
  to a bare string

Assisted-by: Claude Sonnet 4.6
---
 .../messagebroker_kafkapython/test_cluster_metrics_unit.py | 7 ++++---
 tests/messagebroker_kafkapython/test_consumer.py           | 4 +++-
 tests/messagebroker_kafkapython/test_producer.py           | 4 +++-
 3 files changed, 10 insertions(+), 5 deletions(-)

diff --git a/tests/messagebroker_kafkapython/test_cluster_metrics_unit.py b/tests/messagebroker_kafkapython/test_cluster_metrics_unit.py
index e0cce4322d..73d449d597 100644
--- a/tests/messagebroker_kafkapython/test_cluster_metrics_unit.py
+++ b/tests/messagebroker_kafkapython/test_cluster_metrics_unit.py
@@ -54,7 +54,7 @@ def test_message_key_not_overwritten_with_cluster_id_in_cache(self):
         """Key must not be replaced by broker address string when cluster ID cached."""
         cluster_id = "test-cluster-uuid"
         cache_key = "broker1:9092,broker2:9092"
-        _kafka_cluster_id_cache[cache_key] = cluster_id
+        _kafka_cluster_id_cache[cache_key] = (cluster_id, time.monotonic())
 
         wrapped = MagicMock(return_value=MagicMock())
         instance = self._make_producer_instance()
@@ -146,7 +146,8 @@ def test_cluster_id_written_to_cache_on_success(self):
             time.sleep(0.5)
 
         try:
-            assert _kafka_cluster_id_cache.get(cache_key) == "fetched-uuid"
+            cached = _kafka_cluster_id_cache.get(cache_key)
+            assert isinstance(cached, tuple) and cached[0] == "fetched-uuid"
         finally:
             _kafka_cluster_id_cache.pop(cache_key, None)
 
@@ -166,7 +167,7 @@ def test_existing_resolved_entry_not_refetched(self):
         """A fully-resolved cache entry (non-empty) skips the fetch and returns immediately."""
         servers = ["broker-resolved:9092"]
         cache_key = self._cache_key(servers)
-        _kafka_cluster_id_cache[cache_key] = "already-resolved-uuid"
+        _kafka_cluster_id_cache[cache_key] = ("already-resolved-uuid", time.monotonic())
 
         with patch("newrelic.hooks.messagebroker_kafkapython.threading.Thread") as mock_thread:
             _fetch_cluster_id_kafka_python(servers)
diff --git a/tests/messagebroker_kafkapython/test_consumer.py b/tests/messagebroker_kafkapython/test_consumer.py
index ac08760ced..67e0c485c6 100644
--- a/tests/messagebroker_kafkapython/test_consumer.py
+++ b/tests/messagebroker_kafkapython/test_consumer.py
@@ -12,6 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import time
+
 import pytest
 from conftest import cache_kafka_consumer_headers
 from testing_support.fixtures import reset_core_stats_engine, validate_attributes
@@ -199,7 +201,7 @@ def seeded_cluster_id(broker):
 
     cache_key = ",".join(sorted(broker))
     test_cluster_id = "test-cluster-consumer-xyz"
-    _kafka_cluster_id_cache[cache_key] = test_cluster_id
+    _kafka_cluster_id_cache[cache_key] = (test_cluster_id, time.monotonic())
     yield test_cluster_id
     _kafka_cluster_id_cache.pop(cache_key, None)
 
diff --git a/tests/messagebroker_kafkapython/test_producer.py b/tests/messagebroker_kafkapython/test_producer.py
index c28afc6cc8..e41fee3d08 100644
--- a/tests/messagebroker_kafkapython/test_producer.py
+++ b/tests/messagebroker_kafkapython/test_producer.py
@@ -12,6 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import time
+
 import pytest
 from conftest import cache_kafka_producer_headers
 from testing_support.validators.validate_messagebroker_headers import validate_messagebroker_headers
@@ -118,7 +120,7 @@ def seeded_cluster_id(broker):
 
     cache_key = ",".join(sorted(broker))
     test_cluster_id = "test-cluster-abc123"
-    _kafka_cluster_id_cache[cache_key] = test_cluster_id
+    _kafka_cluster_id_cache[cache_key] = (test_cluster_id, time.monotonic())
     yield test_cluster_id
     _kafka_cluster_id_cache.pop(cache_key, None)
 

From 082b2df27bfceebd94436f7c92301455854e6aad Mon Sep 17 00:00:00 2001
From: Pulipelly Shashank Reddy <spulipelly@newrelic.com>
Date: Fri, 3 Jul 2026 13:12:36 +0530
Subject: [PATCH 5/6] feat: gate Kafka cluster metrics behind opt-in config
 setting

Add kafka.cluster_metrics_enabled setting (default: False) controlled by
NEW_RELIC_KAFKA_CLUSTER_METRICS_ENABLED env var. Gate _fetch_cluster_id so
all list_topics() calls and cluster metric recording are disabled unless
explicitly opted in.

Assisted-by: Claude Sonnet 4.6
---
 newrelic/core/config.py                        | 7 +++++++
 newrelic/hooks/messagebroker_confluentkafka.py | 4 ++++
 2 files changed, 11 insertions(+)

diff --git a/newrelic/core/config.py b/newrelic/core/config.py
index f92f66d0d7..d80dafd4ea 100644
--- a/newrelic/core/config.py
+++ b/newrelic/core/config.py
@@ -164,6 +164,10 @@ class GCRuntimeMetricsSettings(Settings):
     enabled = False
 
 
+class KafkaSettings(Settings):
+    cluster_metrics_enabled = False
+
+
 class MemoryRuntimeMetricsSettings(Settings):
     pass
 
@@ -660,6 +664,7 @@ class OpentelemetryTracesSettings(Settings):
 _settings.event_harvest_config.harvest_limits = EventHarvestConfigHarvestLimitSettings()
 _settings.event_loop_visibility = EventLoopVisibilitySettings()
 _settings.gc_runtime_metrics = GCRuntimeMetricsSettings()
+_settings.kafka = KafkaSettings()
 _settings.memory_runtime_pid_metrics = MemoryRuntimeMetricsSettings()
 _settings.heroku = HerokuSettings()
 _settings.infinite_tracing = InfiniteTracingSettings()
@@ -991,6 +996,8 @@ def default_otlp_host(host):
 _settings.gc_runtime_metrics.enabled = _environ_as_bool("NEW_RELIC_GC_RUNTIME_METRICS_ENABLED", default=False)
 _settings.gc_runtime_metrics.top_object_count_limit = 5
 
+_settings.kafka.cluster_metrics_enabled = _environ_as_bool("NEW_RELIC_KAFKA_CLUSTER_METRICS_ENABLED", default=False)
+
 _settings.memory_runtime_pid_metrics.enabled = _environ_as_bool(
     "NEW_RELIC_MEMORY_RUNTIME_PID_METRICS_ENABLED", default=True
 )
diff --git a/newrelic/hooks/messagebroker_confluentkafka.py b/newrelic/hooks/messagebroker_confluentkafka.py
index ff4856e6a9..a8ae296e65 100644
--- a/newrelic/hooks/messagebroker_confluentkafka.py
+++ b/newrelic/hooks/messagebroker_confluentkafka.py
@@ -26,6 +26,7 @@
 from newrelic.api.transaction import current_transaction
 from newrelic.common.object_wrapper import function_wrapper, wrap_function_wrapper
 from newrelic.common.package_version_utils import get_package_version
+from newrelic.core.config import global_settings
 
 _logger = logging.getLogger(__name__)
 
@@ -47,6 +48,9 @@
 
 
 def _fetch_cluster_id(instance):
+    settings = global_settings()
+    if not getattr(getattr(settings, "kafka", None), "cluster_metrics_enabled", False):
+        return
     servers = getattr(instance, "_nr_bootstrap_servers", None)
     # Sort so that equivalent broker sets with different orderings share the same key.
     cache_key = ",".join(sorted(servers)) if servers else None

From c576253939b39072bf4a18d79fcae6526784acf4 Mon Sep 17 00:00:00 2001
From: Pulipelly Shashank Reddy <spulipelly@newrelic.com>
Date: Sat, 4 Jul 2026 19:02:12 +0530
Subject: [PATCH 6/6] test(kafka): split compound assertion for ruff PT018

Assisted-by: Claude Sonnet 4.6
---
 tests/messagebroker_kafkapython/test_cluster_metrics_unit.py | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/tests/messagebroker_kafkapython/test_cluster_metrics_unit.py b/tests/messagebroker_kafkapython/test_cluster_metrics_unit.py
index 73d449d597..a07ef58908 100644
--- a/tests/messagebroker_kafkapython/test_cluster_metrics_unit.py
+++ b/tests/messagebroker_kafkapython/test_cluster_metrics_unit.py
@@ -147,7 +147,8 @@ def test_cluster_id_written_to_cache_on_success(self):
 
         try:
             cached = _kafka_cluster_id_cache.get(cache_key)
-            assert isinstance(cached, tuple) and cached[0] == "fetched-uuid"
+            assert isinstance(cached, tuple)
+            assert cached[0] == "fetched-uuid"
         finally:
             _kafka_cluster_id_cache.pop(cache_key, None)