From 3760932d3edc1aa86d3fee7f0a7e233b325a7420 Mon Sep 17 00:00:00 2001
From: Pulipelly Shashank Reddy <spulipelly@newrelic.com>
Date: Mon, 22 Jun 2026 18:23:32 +0530
Subject: [PATCH 01/14] feat(kafka): add messaging.cluster.id span attribute
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

When a service connects to multiple Kafka clusters, or when the same topic name
exists across environments, Kafka spans carry no information about which cluster
a message came from. messaging.cluster.id is defined in the OTel messaging
semantic conventions for Kafka but was not previously set by this instrumentation.

Adds messaging.cluster.id as a span attribute on producer and consumer spans for
Confluent.Kafka (≥2.0.0). A one-time background AdminClient.DescribeClusterAsync()
call is made per unique bootstrap-servers string; security config (SASL/SSL) is
forwarded so authenticated clusters are supported. The attribute is omitted on
older library versions or if resolution has not yet completed — no overhead on
the hot path.

Assisted-by: Claude Sonnet 4.6 <noreply@anthropic.com>
---
 .../instrumentation/aiokafka/utils.py         | 125 ++++++++++++++
 .../tests/test_utils.py                       | 154 ++++++++++++++++++
 .../confluent_kafka/__init__.py               |  21 +++
 .../instrumentation/confluent_kafka/utils.py  |  72 +++++++-
 .../instrumentation/kafka/__init__.py         |  35 +++-
 .../instrumentation/kafka/utils.py            |  76 +++++++++
 6 files changed, 481 insertions(+), 2 deletions(-)

diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py
index ccfe157129..1fd8a7acdc 100644
--- a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py
+++ b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py
@@ -6,6 +6,7 @@
 import asyncio
 import contextlib
 import json
+import threading
 from logging import getLogger
 from typing import (
     TYPE_CHECKING,
@@ -79,6 +80,93 @@ async def __call__(
 
 _LOG = getLogger(__name__)
 
+# Cluster ID cache: maps normalised broker key → resolved UUID or "" (in-flight sentinel).
+# Kafka cluster IDs are stable for the lifetime of a cluster — no TTL needed.
+_cluster_id_cache: dict[str, str] = {}
+_cluster_id_lock = threading.Lock()
+
+
+def _normalize_bootstrap_servers(servers: str | list[str] | tuple) -> str:
+    """Return a sorted, comma-joined cache key for a bootstrap_servers value."""
+    if isinstance(servers, (list, tuple)):
+        servers = ",".join(str(s) for s in servers)
+    parts = [s.strip() for s in str(servers).split(",") if s.strip()]
+    return ",".join(sorted(parts))
+
+
+def _fetch_cluster_id_background(
+    bootstrap_servers: str | list[str],
+    security_kwargs: dict[str, Any] | None = None,
+) -> None:
+    """Fetch the Kafka cluster UUID in a daemon thread and populate the cache.
+
+    Uses a sentinel ("") to prevent concurrent in-flight fetches for the same
+    broker set.  On failure the sentinel is removed so the next call retries.
+    The first few produce/consume spans may not carry the attribute — accepted
+    trade-off to avoid blocking the hot path.
+    """
+    cache_key = _normalize_bootstrap_servers(bootstrap_servers)
+    if not cache_key:
+        return
+    with _cluster_id_lock:
+        if _cluster_id_cache.get(cache_key) is not None:
+            return  # already resolved or fetch already in-flight
+        _cluster_id_cache[cache_key] = ""  # sentinel
+
+    async def _async_fetch() -> None:
+        admin = None
+        try:
+            from aiokafka.admin import AIOKafkaAdminClient  # noqa: PLC0415
+
+            admin = AIOKafkaAdminClient(
+                bootstrap_servers=bootstrap_servers, **(security_kwargs or {})
+            )
+            await asyncio.wait_for(admin.start(), timeout=10.0)
+            # After start() the client has fetched cluster metadata.
+            cluster_id: str | None = getattr(
+                getattr(getattr(admin, "_client", None), "cluster", None),
+                "cluster_id",
+                None,
+            )
+            if cluster_id:
+                with _cluster_id_lock:
+                    _cluster_id_cache[cache_key] = cluster_id
+            else:
+                _LOG.debug(
+                    "opentelemetry-instrumentation-aiokafka: could not read "
+                    "cluster_id via admin._client.cluster.cluster_id — "
+                    "this private API may differ across aiokafka versions; "
+                    "messaging.cluster.id will be absent"
+                )
+                with _cluster_id_lock:
+                    _cluster_id_cache.pop(cache_key, None)
+        except Exception:  # pylint: disable=broad-except
+            with _cluster_id_lock:
+                _cluster_id_cache.pop(cache_key, None)
+        finally:
+            if admin is not None:
+                try:
+                    await admin.stop()
+                except Exception:  # pylint: disable=broad-except
+                    pass
+
+    def _run() -> None:
+        asyncio.run(_async_fetch())
+
+    threading.Thread(
+        target=_run, daemon=True, name="otel-aiokafka-cluster-id"
+    ).start()
+
+
+def _get_cluster_id(bootstrap_servers: str | list[str]) -> str | None:
+    """Return the cached cluster UUID, or ``None`` if not yet resolved."""
+    cache_key = _normalize_bootstrap_servers(bootstrap_servers)
+    val = _cluster_id_cache.get(cache_key)
+    return val if val else None
+
+
+_MESSAGING_CLUSTER_ID = "messaging.cluster.id"
+
 
 def _extract_bootstrap_servers(
     client: aiokafka.AIOKafkaClient,
@@ -90,6 +178,26 @@ def _extract_client_id(client: aiokafka.AIOKafkaClient) -> str:
     return client._client_id
 
 
+def _extract_security_kwargs(client: aiokafka.AIOKafkaClient) -> dict[str, Any]:
+    """Return SSL/SASL kwargs from an AIOKafkaClient for reuse in AIOKafkaAdminClient."""
+    attrs = (
+        ("_ssl_context", "ssl_context"),
+        ("_security_protocol", "security_protocol"),
+        ("_sasl_mechanism", "sasl_mechanism"),
+        ("_sasl_plain_username", "sasl_plain_username"),
+        ("_sasl_plain_password", "sasl_plain_password"),
+        ("_sasl_kerberos_service_name", "sasl_kerberos_service_name"),
+        ("_sasl_kerberos_domain_name", "sasl_kerberos_domain_name"),
+        ("_sasl_oauth_token_provider", "sasl_oauth_token_provider"),
+    )
+    kwargs: dict[str, Any] = {}
+    for attr, key in attrs:
+        val = getattr(client, attr, None)
+        if val is not None:
+            kwargs[key] = val
+    return kwargs
+
+
 def _extract_consumer_group(
     consumer: aiokafka.AIOKafkaConsumer,
 ) -> str | None:
@@ -259,6 +367,10 @@ def _enrich_base_span(
             messaging_attributes.MESSAGING_KAFKA_MESSAGE_KEY, key
         )
 
+    cluster_id = _get_cluster_id(bootstrap_servers)
+    if cluster_id is not None:
+        span.set_attribute(_MESSAGING_CLUSTER_ID, cluster_id)
+
 
 def _enrich_send_span(
     span: Span,
@@ -357,6 +469,10 @@ def _enrich_getmany_poll_span(
     )
     span.set_attribute(messaging_attributes.MESSAGING_CLIENT_ID, client_id)
 
+    cluster_id = _get_cluster_id(bootstrap_servers)
+    if cluster_id is not None:
+        span.set_attribute(_MESSAGING_CLUSTER_ID, cluster_id)
+
     if consumer_group is not None:
         span.set_attribute(
             messaging_attributes.MESSAGING_CONSUMER_GROUP_NAME, consumer_group
@@ -436,6 +552,9 @@ async def _traced_send(
 
         topic = _extract_send_topic(args, kwargs)
         bootstrap_servers = _extract_bootstrap_servers(instance.client)
+        _fetch_cluster_id_background(
+            bootstrap_servers, _extract_security_kwargs(instance.client)
+        )
         client_id = _extract_client_id(instance.client)
         key = _deserialize_key(_extract_send_key(args, kwargs))
         partition = await _extract_send_partition(instance, args, kwargs)
@@ -520,6 +639,9 @@ async def _traced_getone(
 
         if record:
             bootstrap_servers = _extract_bootstrap_servers(instance._client)
+            _fetch_cluster_id_background(
+                bootstrap_servers, _extract_security_kwargs(instance._client)
+            )
             client_id = _extract_client_id(instance._client)
             consumer_group = _extract_consumer_group(instance)
 
@@ -565,6 +687,9 @@ async def _traced_getmany(
 
         if records:
             bootstrap_servers = _extract_bootstrap_servers(instance._client)
+            _fetch_cluster_id_background(
+                bootstrap_servers, _extract_security_kwargs(instance._client)
+            )
             client_id = _extract_client_id(instance._client)
             consumer_group = _extract_consumer_group(instance)
 
diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py
index 0856d1389d..8c211583c9 100644
--- a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py
+++ b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py
@@ -7,14 +7,20 @@
 
 import aiokafka
 
+import opentelemetry.instrumentation.aiokafka.utils as aiokafka_utils
 from opentelemetry.instrumentation.aiokafka.utils import (
     AIOKafkaContextGetter,
     AIOKafkaContextSetter,
+    _MESSAGING_CLUSTER_ID,
     _aiokafka_getter,
     _aiokafka_setter,
+    _cluster_id_cache,
     _create_consumer_span,
     _extract_send_partition,
+    _fetch_cluster_id_background,
+    _get_cluster_id,
     _get_span_name,
+    _normalize_bootstrap_servers,
     _wrap_getmany,
     _wrap_getone,
     _wrap_send,
@@ -358,6 +364,154 @@ async def test_create_consumer_span(
         )
         detach.assert_called_once_with(attach.return_value)
 
+    async def test_cluster_id_attribute_set_on_send_span(self) -> None:
+        """Cluster ID is added to producer span when cache is pre-populated."""
+        tracer = mock.MagicMock()
+        span = mock.MagicMock()
+        span.is_recording.return_value = True
+        tracer.start_as_current_span.return_value.__enter__ = mock.Mock(
+            return_value=span
+        )
+        tracer.start_as_current_span.return_value.__exit__ = mock.Mock(
+            return_value=False
+        )
+
+        servers = "broker1:9092,broker2:9092"
+        cache_key = _normalize_bootstrap_servers(servers)
+        _cluster_id_cache[cache_key] = "test-cluster-uuid"
+        try:
+            producer = mock.MagicMock()
+            producer.client._bootstrap_servers = servers
+            producer.client._client_id = "test-client"
+            producer.client._wait_on_metadata = mock.AsyncMock()
+            producer._key_serializer = None
+            producer._value_serializer = None
+            producer._partition.return_value = 0
+
+            wrapped_send = _wrap_send(tracer, None)
+            await wrapped_send(
+                mock.AsyncMock(), producer, [self.topic_name], {}
+            )
+
+            set_attribute_calls = {
+                call.args[0]: call.args[1]
+                for call in span.set_attribute.call_args_list
+            }
+            self.assertEqual(
+                set_attribute_calls.get(_MESSAGING_CLUSTER_ID),
+                "test-cluster-uuid",
+            )
+        finally:
+            _cluster_id_cache.pop(cache_key, None)
+
+    async def test_cluster_id_attribute_absent_when_not_resolved(self) -> None:
+        """No cluster ID attribute is set when cache has no entry yet."""
+        tracer = mock.MagicMock()
+        span = mock.MagicMock()
+        span.is_recording.return_value = True
+        tracer.start_as_current_span.return_value.__enter__ = mock.Mock(
+            return_value=span
+        )
+        tracer.start_as_current_span.return_value.__exit__ = mock.Mock(
+            return_value=False
+        )
+
+        servers = "unknown-broker:9092"
+        cache_key = _normalize_bootstrap_servers(servers)
+        _cluster_id_cache.pop(cache_key, None)
+
+        with mock.patch(
+            "opentelemetry.instrumentation.aiokafka.utils._fetch_cluster_id_background"
+        ):
+            producer = mock.MagicMock()
+            producer.client._bootstrap_servers = servers
+            producer.client._client_id = "test-client"
+            producer.client._wait_on_metadata = mock.AsyncMock()
+            producer._key_serializer = None
+            producer._value_serializer = None
+            producer._partition.return_value = 0
+
+            wrapped_send = _wrap_send(tracer, None)
+            await wrapped_send(
+                mock.AsyncMock(), producer, [self.topic_name], {}
+            )
+
+        attribute_keys = [
+            call.args[0] for call in span.set_attribute.call_args_list
+        ]
+        self.assertNotIn(_MESSAGING_CLUSTER_ID, attribute_keys)
+
+    def test_normalize_bootstrap_servers_sorts_and_deduplicates_whitespace(
+        self,
+    ) -> None:
+        self.assertEqual(
+            _normalize_bootstrap_servers("b:9092, a:9092"),
+            "a:9092,b:9092",
+        )
+        self.assertEqual(
+            _normalize_bootstrap_servers(["b:9092", "a:9092"]),
+            "a:9092,b:9092",
+        )
+
+    def test_get_cluster_id_returns_none_for_sentinel(self) -> None:
+        servers = "sentinel-broker:9092"
+        cache_key = _normalize_bootstrap_servers(servers)
+        _cluster_id_cache[cache_key] = ""  # sentinel
+        try:
+            self.assertIsNone(_get_cluster_id(servers))
+        finally:
+            _cluster_id_cache.pop(cache_key, None)
+
+    def test_get_cluster_id_returns_resolved_value(self) -> None:
+        servers = "resolved-broker:9092"
+        cache_key = _normalize_bootstrap_servers(servers)
+        _cluster_id_cache[cache_key] = "abc-uuid"
+        try:
+            self.assertEqual(_get_cluster_id(servers), "abc-uuid")
+        finally:
+            _cluster_id_cache.pop(cache_key, None)
+
+    def test_fetch_cluster_id_background_sets_sentinel_and_spawns_thread(
+        self,
+    ) -> None:
+        servers = "fetch-test-broker:9092"
+        cache_key = _normalize_bootstrap_servers(servers)
+        _cluster_id_cache.pop(cache_key, None)
+
+        with mock.patch("threading.Thread") as mock_thread:
+            mock_thread.return_value.start = mock.Mock()
+            _fetch_cluster_id_background(servers)
+            self.assertEqual(_cluster_id_cache.get(cache_key), "")
+            mock_thread.return_value.start.assert_called_once()
+
+        _cluster_id_cache.pop(cache_key, None)
+
+    def test_fetch_cluster_id_background_skips_if_sentinel_exists(
+        self,
+    ) -> None:
+        servers = "sentinel-skip-broker:9092"
+        cache_key = _normalize_bootstrap_servers(servers)
+        _cluster_id_cache[cache_key] = ""  # pre-set sentinel
+        try:
+            with mock.patch("threading.Thread") as mock_thread:
+                _fetch_cluster_id_background(servers)
+                mock_thread.assert_not_called()
+        finally:
+            _cluster_id_cache.pop(cache_key, None)
+
+    def test_fetch_cluster_id_background_skips_if_already_resolved(
+        self,
+    ) -> None:
+        servers = "resolved-skip-broker:9092"
+        cache_key = _normalize_bootstrap_servers(servers)
+        _cluster_id_cache[cache_key] = "existing-uuid"
+        try:
+            with mock.patch("threading.Thread") as mock_thread:
+                _fetch_cluster_id_background(servers)
+                mock_thread.assert_not_called()
+        finally:
+            _cluster_id_cache.pop(cache_key, None)
+
     async def test_kafka_properties_extractor(self):
         aiokafka_instance_mock = mock.Mock()
         aiokafka_instance_mock._key_serializer = None
diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py
index 08aa4c3c7d..6a6d9cd62d 100644
--- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py
+++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py
@@ -117,6 +117,7 @@ def instrument_consumer(consumer: Consumer, tracer_provider=None): ...
     _create_new_consume_span,
     _end_current_consume_span,
     _enrich_span,
+    _fetch_cluster_id_background,
     _get_span_name,
     _kafka_setter,
 )
@@ -143,6 +144,11 @@ class AutoInstrumentedProducer(Producer):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.config = _capture_config(args, kwargs)
+        bootstrap_servers = KafkaPropertiesExtractor.extract_bootstrap_servers(
+            self
+        )
+        if bootstrap_servers:
+            _fetch_cluster_id_background(bootstrap_servers, self.config)
 
     # This method is deliberately implemented in order to allow wrapt to wrap this function
     def produce(self, topic, value=None, *args, **kwargs):  # pylint: disable=keyword-arg-before-vararg,useless-super-delegation
@@ -154,6 +160,11 @@ def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.config = _capture_config(args, kwargs)
         self._current_consume_span = None
+        bootstrap_servers = KafkaPropertiesExtractor.extract_bootstrap_servers(
+            self
+        )
+        if bootstrap_servers:
+            _fetch_cluster_id_background(bootstrap_servers, self.config)
 
     # This method is deliberately implemented in order to allow wrapt to wrap this function
     def poll(self, timeout=-1):  # pylint: disable=useless-super-delegation
@@ -176,6 +187,11 @@ def __init__(self, producer: Producer, tracer: Tracer):
         # KafkaPropertiesExtractor.extract_bootstrap_servers can read it
         # through this proxy.
         self.config = getattr(producer, "config", None)
+        bootstrap_servers = KafkaPropertiesExtractor.extract_bootstrap_servers(
+            self
+        )
+        if bootstrap_servers:
+            _fetch_cluster_id_background(bootstrap_servers, self.config)
 
     def flush(self, timeout=-1):
         return self._producer.flush(timeout)
@@ -207,6 +223,11 @@ def __init__(self, consumer: Consumer, tracer: Tracer):
         self._current_context_token = None
         # See ProxiedProducer.__init__ for rationale.
         self.config = getattr(consumer, "config", None)
+        bootstrap_servers = KafkaPropertiesExtractor.extract_bootstrap_servers(
+            self
+        )
+        if bootstrap_servers:
+            _fetch_cluster_id_background(bootstrap_servers, self.config)
 
     def close(self, *args, **kwargs):
         return ConfluentKafkaInstrumentor.wrap_close(
diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py
index 27a188e7c7..924635135f 100644
--- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py
+++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py
@@ -1,8 +1,9 @@
 # Copyright The OpenTelemetry Authors
 # SPDX-License-Identifier: Apache-2.0
 
+import threading
 from logging import getLogger
-from typing import List, Optional
+from typing import Dict, List, Optional
 
 from opentelemetry import context, propagate
 from opentelemetry.propagators import textmap
@@ -24,6 +25,71 @@
 
 _LOG = getLogger(__name__)
 
+_MESSAGING_CLUSTER_ID = "messaging.cluster.id"
+
+_kafka_cluster_id_cache: Dict[str, str] = {}
+_kafka_cluster_id_lock = threading.Lock()
+
+
+def _bootstrap_cache_key(bootstrap_servers: Optional[str]) -> str:
+    if not bootstrap_servers:
+        return ""
+    parts = [s.strip() for s in bootstrap_servers.split(",") if s.strip()]
+    return ",".join(sorted(parts))
+
+
+def _fetch_cluster_id_background(
+    bootstrap_servers: Optional[str],
+    base_config: Optional[Dict[str, str]] = None,
+) -> None:
+    """Fetch cluster UUID via confluent_kafka AdminClient in a daemon thread. Caches result by broker key."""
+    if not bootstrap_servers:
+        return
+    cache_key = _bootstrap_cache_key(bootstrap_servers)
+    if not cache_key:
+        return
+    with _kafka_cluster_id_lock:
+        existing = _kafka_cluster_id_cache.get(cache_key)
+        if existing is not None:  # sentinel "" or resolved UUID — skip
+            return
+        _kafka_cluster_id_cache[cache_key] = ""  # sentinel
+
+    def _run() -> None:
+        admin = None
+        try:
+            from confluent_kafka.admin import AdminClient  # noqa: PLC0415
+
+            config = {**(base_config or {}), "bootstrap.servers": bootstrap_servers}
+            admin = AdminClient(config)
+            cluster_metadata = admin.list_topics(timeout=10)
+            cluster_id = cluster_metadata.cluster_id
+            if cluster_id:
+                with _kafka_cluster_id_lock:
+                    _kafka_cluster_id_cache[cache_key] = cluster_id
+            else:
+                with _kafka_cluster_id_lock:
+                    _kafka_cluster_id_cache.pop(cache_key, None)
+        except Exception:  # pylint: disable=broad-except
+            with _kafka_cluster_id_lock:
+                _kafka_cluster_id_cache.pop(cache_key, None)
+        finally:
+            # confluent_kafka.AdminClient has no explicit close(); deleting the reference
+            # allows librdkafka to release native resources via __del__ rather than waiting for GC.
+            del admin
+
+    t = threading.Thread(
+        target=_run, daemon=True, name="otel-confluent-kafka-cluster-id"
+    )
+    t.start()
+
+
+def _get_cluster_id(bootstrap_servers: Optional[str]) -> Optional[str]:
+    if not bootstrap_servers:
+        return None
+    cache_key = _bootstrap_cache_key(bootstrap_servers)
+    val = _kafka_cluster_id_cache.get(cache_key)
+    return val if val else None
+
 
 class KafkaPropertiesExtractor:
     @staticmethod
@@ -183,6 +249,10 @@ def _enrich_span(
 
     _set_bootstrap_servers_attributes(span, bootstrap_servers)
 
+    cluster_id = _get_cluster_id(bootstrap_servers)
+    if cluster_id:
+        span.set_attribute(_MESSAGING_CLUSTER_ID, cluster_id)
+
     # https://stackoverflow.com/questions/65935155/identify-and-find-specific-message-in-kafka-topic
     # A message within Kafka is uniquely defined by its topic name, topic partition and offset.
     if partition is not None and offset is not None and topic:
diff --git a/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/__init__.py b/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/__init__.py
index 16113338fe..96dcd756c9 100644
--- a/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/__init__.py
+++ b/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/__init__.py
@@ -95,7 +95,12 @@ def process_msg(message):
     _instruments_kafka_python,
     _instruments_kafka_python_ng,
 )
-from opentelemetry.instrumentation.kafka.utils import _wrap_next, _wrap_send
+from opentelemetry.instrumentation.kafka.utils import (
+    KafkaPropertiesExtractor,
+    _fetch_cluster_id_background,
+    _wrap_next,
+    _wrap_send,
+)
 from opentelemetry.instrumentation.kafka.version import __version__
 from opentelemetry.instrumentation.utils import unwrap
 
@@ -145,6 +150,32 @@ def _instrument(self, **kwargs):
             schema_url="https://opentelemetry.io/schemas/1.11.0",
         )
 
+        def _wrap_producer_init(func, instance, args, kwargs):
+            func(*args, **kwargs)
+            bootstrap_servers = (
+                KafkaPropertiesExtractor.extract_bootstrap_servers(instance)
+            )
+            if bootstrap_servers:
+                _fetch_cluster_id_background(
+                    bootstrap_servers, getattr(instance, "config", None)
+                )
+
+        def _wrap_consumer_init(func, instance, args, kwargs):
+            func(*args, **kwargs)
+            bootstrap_servers = (
+                KafkaPropertiesExtractor.extract_bootstrap_servers(instance)
+            )
+            if bootstrap_servers:
+                _fetch_cluster_id_background(
+                    bootstrap_servers, getattr(instance, "config", None)
+                )
+
+        wrap_function_wrapper(
+            kafka.KafkaProducer, "__init__", _wrap_producer_init
+        )
+        wrap_function_wrapper(
+            kafka.KafkaConsumer, "__init__", _wrap_consumer_init
+        )
         wrap_function_wrapper(
             kafka.KafkaProducer, "send", _wrap_send(tracer, produce_hook)
         )
@@ -155,5 +186,7 @@ def _instrument(self, **kwargs):
         )
 
     def _uninstrument(self, **kwargs):
+        unwrap(kafka.KafkaProducer, "__init__")
+        unwrap(kafka.KafkaConsumer, "__init__")
         unwrap(kafka.KafkaProducer, "send")
         unwrap(kafka.KafkaConsumer, "__next__")
diff --git a/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/utils.py b/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/utils.py
index 00da544325..8b8df405ed 100644
--- a/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/utils.py
+++ b/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/utils.py
@@ -2,6 +2,7 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import json
+import threading
 from logging import getLogger
 from typing import Callable, Dict, List, Optional
 
@@ -15,6 +16,78 @@
 
 _LOG = getLogger(__name__)
 
+_MESSAGING_CLUSTER_ID = "messaging.cluster.id"
+
+_SECURITY_CONFIG_KEYS = frozenset({
+    "ssl_cafile", "ssl_certfile", "ssl_keyfile", "ssl_password",
+    "ssl_crlfile", "ssl_check_hostname", "ssl_context",
+    "security_protocol", "sasl_mechanism", "sasl_plain_username",
+    "sasl_plain_password", "sasl_kerberos_service_name",
+    "sasl_kerberos_domain_name", "sasl_oauth_token_provider",
+})
+
+_kafka_cluster_id_cache: Dict[str, str] = {}
+_kafka_cluster_id_lock = threading.Lock()
+
+
+def _bootstrap_cache_key(servers) -> str:
+    if isinstance(servers, (list, tuple)):
+        return ",".join(sorted(str(s) for s in servers))
+    parts = [s.strip() for s in str(servers).split(",") if s.strip()]
+    return ",".join(sorted(parts))
+
+
+def _fetch_cluster_id_background(bootstrap_servers, extra_config=None) -> None:
+    """Fetch cluster UUID via KafkaAdminClient in a daemon thread. Caches result by broker key."""
+    cache_key = _bootstrap_cache_key(bootstrap_servers)
+    with _kafka_cluster_id_lock:
+        existing = _kafka_cluster_id_cache.get(cache_key)
+        if existing is not None:  # sentinel "" or resolved UUID — skip
+            return
+        _kafka_cluster_id_cache[cache_key] = ""  # sentinel
+
+    def _run() -> None:
+        admin = None
+        try:
+            from kafka.admin import KafkaAdminClient  # noqa: PLC0415
+
+            security_kwargs = {
+                k: v
+                for k, v in (extra_config or {}).items()
+                if k in _SECURITY_CONFIG_KEYS and v is not None
+            }
+            admin = KafkaAdminClient(
+                bootstrap_servers=bootstrap_servers, **security_kwargs
+            )
+            info = admin.describe_cluster()
+            cluster_id = info.get("cluster_id")
+            if cluster_id:
+                with _kafka_cluster_id_lock:
+                    _kafka_cluster_id_cache[cache_key] = cluster_id
+            else:
+                with _kafka_cluster_id_lock:
+                    _kafka_cluster_id_cache.pop(cache_key, None)
+        except Exception:  # pylint: disable=broad-except
+            with _kafka_cluster_id_lock:
+                _kafka_cluster_id_cache.pop(cache_key, None)
+        finally:
+            if admin is not None:
+                try:
+                    admin.close()
+                except Exception:  # pylint: disable=broad-except
+                    pass
+
+    t = threading.Thread(
+        target=_run, daemon=True, name="otel-kafka-cluster-id"
+    )
+    t.start()
+
+
+def _get_cluster_id(bootstrap_servers) -> Optional[str]:
+    cache_key = _bootstrap_cache_key(bootstrap_servers)
+    val = _kafka_cluster_id_cache.get(cache_key)
+    return val if val else None
+
 
 class KafkaPropertiesExtractor:
     @staticmethod
@@ -135,6 +208,9 @@ def _enrich_span(
         span.set_attribute(
             SpanAttributes.MESSAGING_URL, json.dumps(bootstrap_servers)
         )
+        cluster_id = _get_cluster_id(bootstrap_servers)
+        if cluster_id:
+            span.set_attribute(_MESSAGING_CLUSTER_ID, cluster_id)
 
 
 def _get_span_name(operation: str, topic: str):

From 3f6b0f580ec05f428c9f33f70f8a8060182f67eb Mon Sep 17 00:00:00 2001
From: Pulipelly Shashank Reddy <spulipelly@newrelic.com>
Date: Fri, 26 Jun 2026 20:32:17 +0530
Subject: [PATCH 02/14] feat(kafka): add 30-minute TTL to cluster-id caches

Cache values change from a plain string to (cluster_id, timestamp) tuples.
On read, isinstance(val, tuple) distinguishes a resolved entry from the ""
in-flight sentinel. When the TTL has elapsed the guard falls through to
re-fetch; the stale tuple stays in cache until the background thread stores
a fresh one or removes the key on failure.

Also fix an indentation bug in aiokafka _wrap_send where the return was
outside the span context manager.

Assisted-by: Claude Sonnet 4.6
---
 .../instrumentation/aiokafka/utils.py         |  3 +--
 .../instrumentation/confluent_kafka/utils.py  | 25 +++++++++++++------
 .../instrumentation/kafka/utils.py            | 23 +++++++++++------
 3 files changed, 33 insertions(+), 18 deletions(-)

diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py
index 1fd8a7acdc..8bf399b754 100644
--- a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py
+++ b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py
@@ -580,8 +580,7 @@ async def _traced_send(
                     await async_produce_hook(span, args, kwargs)
             except Exception as hook_exception:  # pylint: disable=W0703
                 _LOG.exception(hook_exception)
-
-        return await func(*args, **kwargs)
+            return await func(*args, **kwargs)
 
     return _traced_send
 
diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py
index 924635135f..8c31eb0bb8 100644
--- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py
+++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py
@@ -2,6 +2,7 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import threading
+import time
 from logging import getLogger
 from typing import Dict, List, Optional
 
@@ -27,7 +28,9 @@
 
 _MESSAGING_CLUSTER_ID = "messaging.cluster.id"
 
-_kafka_cluster_id_cache: Dict[str, str] = {}
+_CLUSTER_ID_TTL_SECONDS = 30 * 60
+
+_kafka_cluster_id_cache: Dict[str, object] = {}
 _kafka_cluster_id_lock = threading.Lock()
 
 
@@ -50,8 +53,12 @@ def _fetch_cluster_id_background(
         return
     with _kafka_cluster_id_lock:
         existing = _kafka_cluster_id_cache.get(cache_key)
-        if existing is not None:  # sentinel "" or resolved UUID — skip
-            return
+        if isinstance(existing, tuple):
+            if time.monotonic() - existing[1] <= _CLUSTER_ID_TTL_SECONDS:
+                return  # still fresh; stale value stays until re-fetch succeeds
+            # TTL expired — fall through to re-fetch
+        elif existing is not None:
+            return  # "" sentinel — fetch already in progress
         _kafka_cluster_id_cache[cache_key] = ""  # sentinel
 
     def _run() -> None:
@@ -65,7 +72,7 @@ def _run() -> None:
             cluster_id = cluster_metadata.cluster_id
             if cluster_id:
                 with _kafka_cluster_id_lock:
-                    _kafka_cluster_id_cache[cache_key] = cluster_id
+                    _kafka_cluster_id_cache[cache_key] = (cluster_id, time.monotonic())
             else:
                 with _kafka_cluster_id_lock:
                     _kafka_cluster_id_cache.pop(cache_key, None)
@@ -80,7 +87,11 @@ def _run() -> None:
     t = threading.Thread(
         target=_run, daemon=True, name="otel-confluent-kafka-cluster-id"
     )
-    t.start()
+    try:
+        t.start()
+    except Exception:  # pylint: disable=broad-except
+        with _kafka_cluster_id_lock:
+            _kafka_cluster_id_cache.pop(cache_key, None)
 
 
 def _get_cluster_id(bootstrap_servers: Optional[str]) -> Optional[str]:
@@ -88,7 +99,7 @@ def _get_cluster_id(bootstrap_servers: Optional[str]) -> Optional[str]:
         return None
     cache_key = _bootstrap_cache_key(bootstrap_servers)
     val = _kafka_cluster_id_cache.get(cache_key)
-    return val if val else None
+    return val[0] if isinstance(val, tuple) else None
 
 
 class KafkaPropertiesExtractor:
@@ -244,8 +255,6 @@ def _enrich_span(
 
     if operation:
         span.set_attribute(MESSAGING_OPERATION, operation.value)
-    else:
-        span.set_attribute(SpanAttributes.MESSAGING_TEMP_DESTINATION, True)
 
     _set_bootstrap_servers_attributes(span, bootstrap_servers)
 
diff --git a/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/utils.py b/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/utils.py
index 8b8df405ed..842b9c7d7d 100644
--- a/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/utils.py
+++ b/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/utils.py
@@ -3,6 +3,7 @@
 
 import json
 import threading
+import time
 from logging import getLogger
 from typing import Callable, Dict, List, Optional
 
@@ -26,7 +27,9 @@
     "sasl_kerberos_domain_name", "sasl_oauth_token_provider",
 })
 
-_kafka_cluster_id_cache: Dict[str, str] = {}
+_CLUSTER_ID_TTL_SECONDS = 30 * 60
+
+_kafka_cluster_id_cache: Dict[str, object] = {}
 _kafka_cluster_id_lock = threading.Lock()
 
 
@@ -42,8 +45,12 @@ def _fetch_cluster_id_background(bootstrap_servers, extra_config=None) -> None:
     cache_key = _bootstrap_cache_key(bootstrap_servers)
     with _kafka_cluster_id_lock:
         existing = _kafka_cluster_id_cache.get(cache_key)
-        if existing is not None:  # sentinel "" or resolved UUID — skip
-            return
+        if isinstance(existing, tuple):
+            if time.monotonic() - existing[1] <= _CLUSTER_ID_TTL_SECONDS:
+                return  # still fresh; stale value stays until re-fetch succeeds
+            # TTL expired — fall through to re-fetch
+        elif existing is not None:
+            return  # "" sentinel — fetch already in progress
         _kafka_cluster_id_cache[cache_key] = ""  # sentinel
 
     def _run() -> None:
@@ -63,7 +70,7 @@ def _run() -> None:
             cluster_id = info.get("cluster_id")
             if cluster_id:
                 with _kafka_cluster_id_lock:
-                    _kafka_cluster_id_cache[cache_key] = cluster_id
+                    _kafka_cluster_id_cache[cache_key] = (cluster_id, time.monotonic())
             else:
                 with _kafka_cluster_id_lock:
                     _kafka_cluster_id_cache.pop(cache_key, None)
@@ -86,7 +93,7 @@ def _run() -> None:
 def _get_cluster_id(bootstrap_servers) -> Optional[str]:
     cache_key = _bootstrap_cache_key(bootstrap_servers)
     val = _kafka_cluster_id_cache.get(cache_key)
-    return val if val else None
+    return val[0] if isinstance(val, tuple) else None
 
 
 class KafkaPropertiesExtractor:
@@ -232,6 +239,7 @@ def _traced_send(func, instance, args, kwargs):
             instance, args, kwargs
         )
         span_name = _get_span_name("send", topic)
+        _fetch_cluster_id_background(bootstrap_servers, dict(instance.config))
         with tracer.start_as_current_span(
             span_name, kind=trace.SpanKind.PRODUCER
         ) as span:
@@ -246,8 +254,7 @@ def _traced_send(func, instance, args, kwargs):
                     produce_hook(span, args, kwargs)
             except Exception as hook_exception:  # pylint: disable=W0703
                 _LOG.exception(hook_exception)
-
-        return func(*args, **kwargs)
+            return func(*args, **kwargs)
 
     return _traced_send
 
@@ -290,7 +297,7 @@ def _traced_next(func, instance, args, kwargs):
             bootstrap_servers = (
                 KafkaPropertiesExtractor.extract_bootstrap_servers(instance)
             )
-
+            _fetch_cluster_id_background(bootstrap_servers, dict(instance.config))
             extracted_context = propagate.extract(
                 record.headers, getter=_kafka_getter
             )

From 25bad0a0619ac99124ac254672156693a4970960 Mon Sep 17 00:00:00 2001
From: Pulipelly Shashank Reddy <spulipelly@newrelic.com>
Date: Fri, 26 Jun 2026 22:18:13 +0530
Subject: [PATCH 03/14] fix(kafka): sentinel leak on thread failure; TTL
 enforcement on hot path

kafka-python (utils.py):
  t.start() was not guarded by try/except. If the OS failed to create the
  thread, the "" sentinel would remain in _kafka_cluster_id_cache permanently,
  blocking all future fetches for that broker key. Fix: wrap t.start() in
  try/except; remove the sentinel on failure (matching the confluent-kafka
  pattern already in place).

confluent-kafka (utils.py):
  _fetch_cluster_id_background was called from producer/consumer constructor
  interceptors but not from _enrich_span (the hot path). Long-lived clients
  never triggered TTL re-fetches after the initial fetch expired.
  Fix: call _fetch_cluster_id_background in _enrich_span before _get_cluster_id.
  The function is a no-op when the cached value is still fresh (TTL not yet
  expired), and starts a background re-fetch only when the TTL has lapsed.

  To preserve auth credentials for SSL/SASL clusters during TTL re-fetches:
  _fetch_cluster_id_background now stores the base_config from the first call
  (constructor path, which has the full producer/consumer config) in a separate
  _kafka_cluster_id_config_cache dict keyed by bootstrap_servers. When called
  from _enrich_span without credentials, it looks up the stored config and
  passes it to the AdminClient, so re-fetches succeed on authenticated clusters.

Assisted-by: Claude Sonnet 4.6
---
 .../instrumentation/confluent_kafka/utils.py        | 13 ++++++++++++-
 .../opentelemetry/instrumentation/kafka/utils.py    |  6 +++++-
 2 files changed, 17 insertions(+), 2 deletions(-)

diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py
index 8c31eb0bb8..599dfa9ed7 100644
--- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py
+++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py
@@ -32,6 +32,8 @@
 
 _kafka_cluster_id_cache: Dict[str, object] = {}
 _kafka_cluster_id_lock = threading.Lock()
+# Auth config stored from the first fetch per broker key; used for TTL re-fetches.
+_kafka_cluster_id_config_cache: Dict[str, Optional[Dict[str, str]]] = {}
 
 
 def _bootstrap_cache_key(bootstrap_servers: Optional[str]) -> str:
@@ -51,6 +53,13 @@ def _fetch_cluster_id_background(
     cache_key = _bootstrap_cache_key(bootstrap_servers)
     if not cache_key:
         return
+
+    # Store auth config from the first call (constructor) so TTL re-fetches without
+    # credentials (hot path) can still connect to SSL/SASL-protected clusters.
+    if base_config is not None:
+        _kafka_cluster_id_config_cache.setdefault(cache_key, base_config)
+    resolved_config = _kafka_cluster_id_config_cache.get(cache_key) or base_config
+
     with _kafka_cluster_id_lock:
         existing = _kafka_cluster_id_cache.get(cache_key)
         if isinstance(existing, tuple):
@@ -66,7 +75,7 @@ def _run() -> None:
         try:
             from confluent_kafka.admin import AdminClient  # noqa: PLC0415
 
-            config = {**(base_config or {}), "bootstrap.servers": bootstrap_servers}
+            config = {**(resolved_config or {}), "bootstrap.servers": bootstrap_servers}
             admin = AdminClient(config)
             cluster_metadata = admin.list_topics(timeout=10)
             cluster_id = cluster_metadata.cluster_id
@@ -258,6 +267,8 @@ def _enrich_span(
 
     _set_bootstrap_servers_attributes(span, bootstrap_servers)
 
+    # Trigger a background re-fetch if the cached value has expired (TTL enforcement).
+    _fetch_cluster_id_background(bootstrap_servers)
     cluster_id = _get_cluster_id(bootstrap_servers)
     if cluster_id:
         span.set_attribute(_MESSAGING_CLUSTER_ID, cluster_id)
diff --git a/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/utils.py b/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/utils.py
index 842b9c7d7d..8958c75444 100644
--- a/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/utils.py
+++ b/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/utils.py
@@ -87,7 +87,11 @@ def _run() -> None:
     t = threading.Thread(
         target=_run, daemon=True, name="otel-kafka-cluster-id"
     )
-    t.start()
+    try:
+        t.start()
+    except Exception:  # pylint: disable=broad-except
+        with _kafka_cluster_id_lock:
+            _kafka_cluster_id_cache.pop(cache_key, None)
 
 
 def _get_cluster_id(bootstrap_servers) -> Optional[str]:

From 4431d1ac47ef720bbf720869b583de5e06fd041b Mon Sep 17 00:00:00 2001
From: Pulipelly Shashank Reddy <spulipelly@newrelic.com>
Date: Sun, 28 Jun 2026 13:58:09 +0530
Subject: [PATCH 04/14] chore(kafka): increase cluster ID cache TTL from 30 min
 to 1 hour

Assisted-by: Claude Sonnet 4.6
---
 .../src/opentelemetry/instrumentation/kafka/utils.py            | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/utils.py b/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/utils.py
index 8958c75444..4a83c2ce82 100644
--- a/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/utils.py
+++ b/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/utils.py
@@ -27,7 +27,7 @@
     "sasl_kerberos_domain_name", "sasl_oauth_token_provider",
 })
 
-_CLUSTER_ID_TTL_SECONDS = 30 * 60
+_CLUSTER_ID_TTL_SECONDS = 60 * 60
 
 _kafka_cluster_id_cache: Dict[str, object] = {}
 _kafka_cluster_id_lock = threading.Lock()

From 5c8f6e9dcfd8ea482efa9bb203dc9269485dad45 Mon Sep 17 00:00:00 2001
From: Pulipelly Shashank Reddy <spulipelly@newrelic.com>
Date: Sun, 28 Jun 2026 19:49:58 +0530
Subject: [PATCH 05/14] fix(kafka): increase cluster-id cache TTL from 30 min
 to 60 min

Assisted-by: Claude Sonnet 4.6
---
 .../src/opentelemetry/instrumentation/confluent_kafka/utils.py  | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py
index 599dfa9ed7..f1c51e3309 100644
--- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py
+++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py
@@ -28,7 +28,7 @@
 
 _MESSAGING_CLUSTER_ID = "messaging.cluster.id"
 
-_CLUSTER_ID_TTL_SECONDS = 30 * 60
+_CLUSTER_ID_TTL_SECONDS = 60 * 60
 
 _kafka_cluster_id_cache: Dict[str, object] = {}
 _kafka_cluster_id_lock = threading.Lock()

From 8dade9d2b9d8e47a9d9887264ea8513485a1aa42 Mon Sep 17 00:00:00 2001
From: Pulipelly Shashank Reddy <spulipelly@newrelic.com>
Date: Sun, 28 Jun 2026 23:04:27 +0530
Subject: [PATCH 06/14] confluent-kafka: fetch cluster ID via
 instance.list_topics(), not AdminClient

Replace the separate AdminClient connection used to retrieve the Kafka
cluster UUID with a direct list_topics() call on the existing
Producer/Consumer instance, matching the approach already taken by the
NR Python agent.

- Add _get_real_instance() helper to unwrap ProxiedProducer/Consumer
- _fetch_cluster_id_background() accepts optional `instance` arg; uses
  instance.list_topics() when provided, falls back to AdminClient
- _enrich_span() accepts and forwards the instance
- AutoInstrumentedProducer/Consumer pass instance=self at init
- ProxiedProducer/Consumer pass instance=producer/consumer at init
- wrap_produce/poll/consume pass instance=_get_real_instance(instance)

Assisted-by: Claude Sonnet 4.6
---
 .../confluent_kafka/__init__.py               | 12 ++++--
 .../instrumentation/confluent_kafka/utils.py  | 39 ++++++++++++-------
 2 files changed, 33 insertions(+), 18 deletions(-)

diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py
index 6a6d9cd62d..c45ae6ce64 100644
--- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py
+++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py
@@ -118,6 +118,7 @@ def instrument_consumer(consumer: Consumer, tracer_provider=None): ...
     _end_current_consume_span,
     _enrich_span,
     _fetch_cluster_id_background,
+    _get_real_instance,
     _get_span_name,
     _kafka_setter,
 )
@@ -148,7 +149,7 @@ def __init__(self, *args, **kwargs):
             self
         )
         if bootstrap_servers:
-            _fetch_cluster_id_background(bootstrap_servers, self.config)
+            _fetch_cluster_id_background(bootstrap_servers, self.config, instance=self)
 
     # This method is deliberately implemented in order to allow wrapt to wrap this function
     def produce(self, topic, value=None, *args, **kwargs):  # pylint: disable=keyword-arg-before-vararg,useless-super-delegation
@@ -164,7 +165,7 @@ def __init__(self, *args, **kwargs):
             self
         )
         if bootstrap_servers:
-            _fetch_cluster_id_background(bootstrap_servers, self.config)
+            _fetch_cluster_id_background(bootstrap_servers, self.config, instance=self)
 
     # This method is deliberately implemented in order to allow wrapt to wrap this function
     def poll(self, timeout=-1):  # pylint: disable=useless-super-delegation
@@ -191,7 +192,7 @@ def __init__(self, producer: Producer, tracer: Tracer):
             self
         )
         if bootstrap_servers:
-            _fetch_cluster_id_background(bootstrap_servers, self.config)
+            _fetch_cluster_id_background(bootstrap_servers, self.config, instance=producer)
 
     def flush(self, timeout=-1):
         return self._producer.flush(timeout)
@@ -227,7 +228,7 @@ def __init__(self, consumer: Consumer, tracer: Tracer):
             self
         )
         if bootstrap_servers:
-            _fetch_cluster_id_background(bootstrap_servers, self.config)
+            _fetch_cluster_id_background(bootstrap_servers, self.config, instance=consumer)
 
     def close(self, *args, **kwargs):
         return ConfluentKafkaInstrumentor.wrap_close(
@@ -418,6 +419,7 @@ def wrap_produce(func, instance, tracer, args, kwargs):
                 topic,
                 operation=MessagingOperationTypeValues.PUBLISH,
                 bootstrap_servers=bootstrap_servers,
+                instance=_get_real_instance(instance),
             )  # Publish
             propagate.inject(
                 headers,
@@ -446,6 +448,7 @@ def wrap_poll(func, instance, tracer, args, kwargs):
                     record.offset(),
                     operation=MessagingOperationTypeValues.PROCESS,
                     bootstrap_servers=bootstrap_servers,
+                    instance=_get_real_instance(instance),
                 )
             instance._current_context_token = context.attach(
                 trace.set_span_in_context(instance._current_consume_span)
@@ -472,6 +475,7 @@ def wrap_consume(func, instance, tracer, args, kwargs):
                     records[0].topic(),
                     operation=MessagingOperationTypeValues.PROCESS,
                     bootstrap_servers=bootstrap_servers,
+                    instance=_get_real_instance(instance),
                 )
             instance._current_context_token = context.attach(
                 trace.set_span_in_context(instance._current_consume_span)
diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py
index f1c51e3309..07a6e1e5cd 100644
--- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py
+++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py
@@ -4,7 +4,7 @@
 import threading
 import time
 from logging import getLogger
-from typing import Dict, List, Optional
+from typing import Any, Dict, List, Optional
 
 from opentelemetry import context, propagate
 from opentelemetry.propagators import textmap
@@ -36,6 +36,11 @@
 _kafka_cluster_id_config_cache: Dict[str, Optional[Dict[str, str]]] = {}
 
 
+def _get_real_instance(instance: Any) -> Any:
+    """Unwrap Proxied* wrappers to get the underlying confluent-kafka Producer/Consumer."""
+    return getattr(instance, "_producer", None) or getattr(instance, "_consumer", None) or instance
+
+
 def _bootstrap_cache_key(bootstrap_servers: Optional[str]) -> str:
     if not bootstrap_servers:
         return ""
@@ -46,8 +51,9 @@ def _bootstrap_cache_key(bootstrap_servers: Optional[str]) -> str:
 def _fetch_cluster_id_background(
     bootstrap_servers: Optional[str],
     base_config: Optional[Dict[str, str]] = None,
+    instance: Optional[Any] = None,
 ) -> None:
-    """Fetch cluster UUID via confluent_kafka AdminClient in a daemon thread. Caches result by broker key."""
+    """Fetch cluster UUID in a daemon thread. Uses instance.list_topics() when available; falls back to AdminClient."""
     if not bootstrap_servers:
         return
     cache_key = _bootstrap_cache_key(bootstrap_servers)
@@ -71,14 +77,22 @@ def _fetch_cluster_id_background(
         _kafka_cluster_id_cache[cache_key] = ""  # sentinel
 
     def _run() -> None:
-        admin = None
         try:
-            from confluent_kafka.admin import AdminClient  # noqa: PLC0415
-
-            config = {**(resolved_config or {}), "bootstrap.servers": bootstrap_servers}
-            admin = AdminClient(config)
-            cluster_metadata = admin.list_topics(timeout=10)
-            cluster_id = cluster_metadata.cluster_id
+            if instance is not None:
+                cluster_metadata = instance.list_topics(timeout=10)
+            else:
+                from confluent_kafka.admin import AdminClient  # noqa: PLC0415
+
+                admin = AdminClient(
+                    {**(resolved_config or {}), "bootstrap.servers": bootstrap_servers}
+                )
+                try:
+                    cluster_metadata = admin.list_topics(timeout=10)
+                finally:
+                    # confluent_kafka.AdminClient has no explicit close(); deleting the reference
+                    # allows librdkafka to release native resources via __del__ rather than waiting for GC.
+                    del admin
+            cluster_id = getattr(cluster_metadata, "cluster_id", None)
             if cluster_id:
                 with _kafka_cluster_id_lock:
                     _kafka_cluster_id_cache[cache_key] = (cluster_id, time.monotonic())
@@ -88,10 +102,6 @@ def _run() -> None:
         except Exception:  # pylint: disable=broad-except
             with _kafka_cluster_id_lock:
                 _kafka_cluster_id_cache.pop(cache_key, None)
-        finally:
-            # confluent_kafka.AdminClient has no explicit close(); deleting the reference
-            # allows librdkafka to release native resources via __del__ rather than waiting for GC.
-            del admin
 
     t = threading.Thread(
         target=_run, daemon=True, name="otel-confluent-kafka-cluster-id"
@@ -247,6 +257,7 @@ def _enrich_span(
     offset: Optional[int] = None,
     operation: Optional[MessagingOperationTypeValues] = None,
     bootstrap_servers: Optional[str] = None,
+    instance: Optional[Any] = None,
 ):
     if not span.is_recording():
         return
@@ -268,7 +279,7 @@ def _enrich_span(
     _set_bootstrap_servers_attributes(span, bootstrap_servers)
 
     # Trigger a background re-fetch if the cached value has expired (TTL enforcement).
-    _fetch_cluster_id_background(bootstrap_servers)
+    _fetch_cluster_id_background(bootstrap_servers, instance=instance)
     cluster_id = _get_cluster_id(bootstrap_servers)
     if cluster_id:
         span.set_attribute(_MESSAGING_CLUSTER_ID, cluster_id)

From 39e87a8231c4dfe787cb68b216b3e822a4c12685 Mon Sep 17 00:00:00 2001
From: Pulipelly Shashank Reddy <spulipelly@newrelic.com>
Date: Sun, 28 Jun 2026 23:05:55 +0530
Subject: [PATCH 07/14] confluent-kafka: protect config cache write with lock

Move _kafka_cluster_id_config_cache.setdefault into the existing
_kafka_cluster_id_lock block to prevent a data race under free-threaded
Python (PEP 703).

Assisted-by: Claude Sonnet 4.6
---
 .../instrumentation/confluent_kafka/utils.py         | 12 ++++++------
 1 file changed, 6 insertions(+), 6 deletions(-)

diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py
index 07a6e1e5cd..18fae3ace5 100644
--- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py
+++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py
@@ -60,13 +60,13 @@ def _fetch_cluster_id_background(
     if not cache_key:
         return
 
-    # Store auth config from the first call (constructor) so TTL re-fetches without
-    # credentials (hot path) can still connect to SSL/SASL-protected clusters.
-    if base_config is not None:
-        _kafka_cluster_id_config_cache.setdefault(cache_key, base_config)
-    resolved_config = _kafka_cluster_id_config_cache.get(cache_key) or base_config
-
     with _kafka_cluster_id_lock:
+        # Store auth config from the first call (constructor) so TTL re-fetches without
+        # credentials (hot path) can still connect to SSL/SASL-protected clusters.
+        if base_config is not None:
+            _kafka_cluster_id_config_cache.setdefault(cache_key, base_config)
+        resolved_config = _kafka_cluster_id_config_cache.get(cache_key) or base_config
+
         existing = _kafka_cluster_id_cache.get(cache_key)
         if isinstance(existing, tuple):
             if time.monotonic() - existing[1] <= _CLUSTER_ID_TTL_SECONDS:

From 63a74229c6395e93030fbb0b60ea324af03aaa40 Mon Sep 17 00:00:00 2001
From: Pulipelly Shashank Reddy <spulipelly@newrelic.com>
Date: Sat, 4 Jul 2026 15:37:10 +0530
Subject: [PATCH 08/14] aiokafka: replace AdminClient cluster-id fetch with
 direct client metadata read
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Remove the background-thread AIOKafkaAdminClient machinery (_fetch_cluster_id_background,
_extract_security_kwargs, _normalize_bootstrap_servers, _get_cluster_id, and the
shared cache/lock). Replace with _extract_cluster_id_from_client(), which reads
AIOKafkaClient.cluster.cluster_id directly — this value is populated by the existing
producer/consumer client after its first broker metadata response, so no extra
connection, thread, or asyncio.run() is needed.

For the send path the attribute is set once at span start (may be None on the very
first send) and again after await func() returns (guaranteed populated after the
broker responds). Consumer spans are created after func() returns so cluster_id is
always available at creation time.

Assisted-by: Claude Sonnet 4.6
---
 .../instrumentation/aiokafka/utils.py         | 157 ++++------------
 .../tests/test_utils.py                       | 177 +++++++-----------
 2 files changed, 106 insertions(+), 228 deletions(-)

diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py
index 8bf399b754..54f54f78be 100644
--- a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py
+++ b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py
@@ -6,7 +6,6 @@
 import asyncio
 import contextlib
 import json
-import threading
 from logging import getLogger
 from typing import (
     TYPE_CHECKING,
@@ -80,91 +79,6 @@ async def __call__(
 
 _LOG = getLogger(__name__)
 
-# Cluster ID cache: maps normalised broker key → resolved UUID or "" (in-flight sentinel).
-# Kafka cluster IDs are stable for the lifetime of a cluster — no TTL needed.
-_cluster_id_cache: dict[str, str] = {}
-_cluster_id_lock = threading.Lock()
-
-
-def _normalize_bootstrap_servers(servers: str | list[str] | tuple) -> str:
-    """Return a sorted, comma-joined cache key for a bootstrap_servers value."""
-    if isinstance(servers, (list, tuple)):
-        servers = ",".join(str(s) for s in servers)
-    parts = [s.strip() for s in str(servers).split(",") if s.strip()]
-    return ",".join(sorted(parts))
-
-
-def _fetch_cluster_id_background(
-    bootstrap_servers: str | list[str],
-    security_kwargs: dict[str, Any] | None = None,
-) -> None:
-    """Fetch the Kafka cluster UUID in a daemon thread and populate the cache.
-
-    Uses a sentinel ("") to prevent concurrent in-flight fetches for the same
-    broker set.  On failure the sentinel is removed so the next call retries.
-    The first few produce/consume spans may not carry the attribute — accepted
-    trade-off to avoid blocking the hot path.
-    """
-    cache_key = _normalize_bootstrap_servers(bootstrap_servers)
-    if not cache_key:
-        return
-    with _cluster_id_lock:
-        if _cluster_id_cache.get(cache_key) is not None:
-            return  # already resolved or fetch already in-flight
-        _cluster_id_cache[cache_key] = ""  # sentinel
-
-    async def _async_fetch() -> None:
-        admin = None
-        try:
-            from aiokafka.admin import AIOKafkaAdminClient  # noqa: PLC0415
-
-            admin = AIOKafkaAdminClient(
-                bootstrap_servers=bootstrap_servers, **(security_kwargs or {})
-            )
-            await asyncio.wait_for(admin.start(), timeout=10.0)
-            # After start() the client has fetched cluster metadata.
-            cluster_id: str | None = getattr(
-                getattr(getattr(admin, "_client", None), "cluster", None),
-                "cluster_id",
-                None,
-            )
-            if cluster_id:
-                with _cluster_id_lock:
-                    _cluster_id_cache[cache_key] = cluster_id
-            else:
-                _LOG.debug(
-                    "opentelemetry-instrumentation-aiokafka: could not read "
-                    "cluster_id via admin._client.cluster.cluster_id — "
-                    "this private API may differ across aiokafka versions; "
-                    "messaging.cluster.id will be absent"
-                )
-                with _cluster_id_lock:
-                    _cluster_id_cache.pop(cache_key, None)
-        except Exception:  # pylint: disable=broad-except
-            with _cluster_id_lock:
-                _cluster_id_cache.pop(cache_key, None)
-        finally:
-            if admin is not None:
-                try:
-                    await admin.stop()
-                except Exception:  # pylint: disable=broad-except
-                    pass
-
-    def _run() -> None:
-        asyncio.run(_async_fetch())
-
-    threading.Thread(
-        target=_run, daemon=True, name="otel-aiokafka-cluster-id"
-    ).start()
-
-
-def _get_cluster_id(bootstrap_servers: str | list[str]) -> str | None:
-    """Return the cached cluster UUID, or ``None`` if not yet resolved."""
-    cache_key = _normalize_bootstrap_servers(bootstrap_servers)
-    val = _cluster_id_cache.get(cache_key)
-    return val if val else None
-
-
 _MESSAGING_CLUSTER_ID = "messaging.cluster.id"
 
 
@@ -178,24 +92,22 @@ def _extract_client_id(client: aiokafka.AIOKafkaClient) -> str:
     return client._client_id
 
 
-def _extract_security_kwargs(client: aiokafka.AIOKafkaClient) -> dict[str, Any]:
-    """Return SSL/SASL kwargs from an AIOKafkaClient for reuse in AIOKafkaAdminClient."""
-    attrs = (
-        ("_ssl_context", "ssl_context"),
-        ("_security_protocol", "security_protocol"),
-        ("_sasl_mechanism", "sasl_mechanism"),
-        ("_sasl_plain_username", "sasl_plain_username"),
-        ("_sasl_plain_password", "sasl_plain_password"),
-        ("_sasl_kerberos_service_name", "sasl_kerberos_service_name"),
-        ("_sasl_kerberos_domain_name", "sasl_kerberos_domain_name"),
-        ("_sasl_oauth_token_provider", "sasl_oauth_token_provider"),
-    )
-    kwargs: dict[str, Any] = {}
-    for attr, key in attrs:
-        val = getattr(client, attr, None)
-        if val is not None:
-            kwargs[key] = val
-    return kwargs
+def _extract_cluster_id_from_client(
+    client: aiokafka.AIOKafkaClient,
+) -> str | None:
+    """Read cluster ID from the aiokafka client's cached cluster metadata.
+
+    aiokafka sets AIOKafkaClient.cluster.cluster_id after the first successful
+    broker metadata response — no extra connection or background thread needed.
+    Returns None if metadata has not been received yet.
+    """
+    try:
+        cluster_id = getattr(
+            getattr(client, "cluster", None), "cluster_id", None
+        )
+        return cluster_id if cluster_id else None
+    except Exception:  # pylint: disable=broad-except
+        return None
 
 
 def _extract_consumer_group(
@@ -345,6 +257,7 @@ def _enrich_base_span(
     topic: str,
     partition: int | None,
     key: str | None,
+    cluster_id: str | None = None,
 ) -> None:
     span.set_attribute(
         messaging_attributes.MESSAGING_SYSTEM,
@@ -367,7 +280,6 @@ def _enrich_base_span(
             messaging_attributes.MESSAGING_KAFKA_MESSAGE_KEY, key
         )
 
-    cluster_id = _get_cluster_id(bootstrap_servers)
     if cluster_id is not None:
         span.set_attribute(_MESSAGING_CLUSTER_ID, cluster_id)
 
@@ -380,6 +292,7 @@ def _enrich_send_span(
     topic: str,
     partition: int | None,
     key: str | None,
+    cluster_id: str | None = None,
 ) -> None:
     if not span.is_recording():
         return
@@ -391,6 +304,7 @@ def _enrich_send_span(
         topic=topic,
         partition=partition,
         key=key,
+        cluster_id=cluster_id,
     )
 
     span.set_attribute(messaging_attributes.MESSAGING_OPERATION_NAME, "send")
@@ -410,6 +324,7 @@ def _enrich_getone_span(
     partition: int | None,
     key: str | None,
     offset: int,
+    cluster_id: str | None = None,
 ) -> None:
     if not span.is_recording():
         return
@@ -421,6 +336,7 @@ def _enrich_getone_span(
         topic=topic,
         partition=partition,
         key=key,
+        cluster_id=cluster_id,
     )
 
     if consumer_group is not None:
@@ -456,6 +372,7 @@ def _enrich_getmany_poll_span(
     client_id: str,
     consumer_group: str | None,
     message_count: int,
+    cluster_id: str | None = None,
 ) -> None:
     if not span.is_recording():
         return
@@ -469,7 +386,6 @@ def _enrich_getmany_poll_span(
     )
     span.set_attribute(messaging_attributes.MESSAGING_CLIENT_ID, client_id)
 
-    cluster_id = _get_cluster_id(bootstrap_servers)
     if cluster_id is not None:
         span.set_attribute(_MESSAGING_CLUSTER_ID, cluster_id)
 
@@ -500,6 +416,7 @@ def _enrich_getmany_topic_span(
     topic: str,
     partition: int,
     message_count: int,
+    cluster_id: str | None = None,
 ) -> None:
     if not span.is_recording():
         return
@@ -511,6 +428,7 @@ def _enrich_getmany_topic_span(
         topic=topic,
         partition=partition,
         key=None,
+        cluster_id=cluster_id,
     )
 
     if consumer_group is not None:
@@ -552,9 +470,6 @@ async def _traced_send(
 
         topic = _extract_send_topic(args, kwargs)
         bootstrap_servers = _extract_bootstrap_servers(instance.client)
-        _fetch_cluster_id_background(
-            bootstrap_servers, _extract_security_kwargs(instance.client)
-        )
         client_id = _extract_client_id(instance.client)
         key = _deserialize_key(_extract_send_key(args, kwargs))
         partition = await _extract_send_partition(instance, args, kwargs)
@@ -569,6 +484,7 @@ async def _traced_send(
                 topic=topic,
                 partition=partition,
                 key=key,
+                cluster_id=_extract_cluster_id_from_client(instance.client),
             )
             propagate.inject(
                 headers,
@@ -580,7 +496,14 @@ async def _traced_send(
                     await async_produce_hook(span, args, kwargs)
             except Exception as hook_exception:  # pylint: disable=W0703
                 _LOG.exception(hook_exception)
-            return await func(*args, **kwargs)
+            result = await func(*args, **kwargs)
+            # After send() returns the broker has responded and cluster metadata
+            # is populated.  Set the attribute now if it wasn't available at span
+            # start (first send on a freshly-created producer).
+            cluster_id = _extract_cluster_id_from_client(instance.client)
+            if cluster_id is not None and span.is_recording():
+                span.set_attribute(_MESSAGING_CLUSTER_ID, cluster_id)
+            return result
 
     return _traced_send
 
@@ -593,6 +516,7 @@ async def _create_consumer_span(
     bootstrap_servers: str | list[str],
     client_id: str,
     consumer_group: str | None,
+    cluster_id: str | None,
     args: tuple[aiokafka.TopicPartition, ...],
     kwargs: dict[str, Any],
 ) -> trace.Span:
@@ -613,6 +537,7 @@ async def _create_consumer_span(
             partition=record.partition,
             key=_deserialize_key(record.key),
             offset=record.offset,
+            cluster_id=cluster_id,
         )
         try:
             if async_consume_hook is not None:
@@ -638,11 +563,9 @@ async def _traced_getone(
 
         if record:
             bootstrap_servers = _extract_bootstrap_servers(instance._client)
-            _fetch_cluster_id_background(
-                bootstrap_servers, _extract_security_kwargs(instance._client)
-            )
             client_id = _extract_client_id(instance._client)
             consumer_group = _extract_consumer_group(instance)
+            cluster_id = _extract_cluster_id_from_client(instance._client)
 
             extracted_context = propagate.extract(
                 record.headers, getter=_aiokafka_getter
@@ -655,6 +578,7 @@ async def _traced_getone(
                 bootstrap_servers,
                 client_id,
                 consumer_group,
+                cluster_id,
                 args,
                 kwargs,
             )
@@ -686,11 +610,9 @@ async def _traced_getmany(
 
         if records:
             bootstrap_servers = _extract_bootstrap_servers(instance._client)
-            _fetch_cluster_id_background(
-                bootstrap_servers, _extract_security_kwargs(instance._client)
-            )
             client_id = _extract_client_id(instance._client)
             consumer_group = _extract_consumer_group(instance)
+            cluster_id = _extract_cluster_id_from_client(instance._client)
 
             span_name = _get_span_name(
                 "receive",
@@ -705,6 +627,7 @@ async def _traced_getmany(
                     client_id=client_id,
                     consumer_group=consumer_group,
                     message_count=sum(len(r) for r in records.values()),
+                    cluster_id=cluster_id,
                 )
 
                 for topic, topic_records in records.items():
@@ -720,6 +643,7 @@ async def _traced_getmany(
                             topic=topic.topic,
                             partition=topic.partition,
                             message_count=len(topic_records),
+                            cluster_id=cluster_id,
                         )
 
                         for record in topic_records:
@@ -734,6 +658,7 @@ async def _traced_getmany(
                                 bootstrap_servers,
                                 client_id,
                                 consumer_group,
+                                cluster_id,
                                 args,
                                 kwargs,
                             )
diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py
index 8c211583c9..53a005f215 100644
--- a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py
+++ b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py
@@ -14,13 +14,10 @@
     _MESSAGING_CLUSTER_ID,
     _aiokafka_getter,
     _aiokafka_setter,
-    _cluster_id_cache,
     _create_consumer_span,
+    _extract_cluster_id_from_client,
     _extract_send_partition,
-    _fetch_cluster_id_background,
-    _get_cluster_id,
     _get_span_name,
-    _normalize_bootstrap_servers,
     _wrap_getmany,
     _wrap_getone,
     _wrap_send,
@@ -145,6 +142,7 @@ async def wrap_send_helper(
             topic=self.topic_name,
             partition=extract_send_partition.return_value,
             key=None,
+            cluster_id=kafka_producer.client.cluster.cluster_id,
         )
 
         set_span_in_context.assert_called_once_with(span)
@@ -220,6 +218,7 @@ async def test_wrap_getone(
             bootstrap_servers,
             client_id,
             consumer_group,
+            kafka_consumer._client.cluster.cluster_id,
             self.args,
             self.kwargs,
         )
@@ -301,6 +300,7 @@ async def test_wrap_getmany(
             bootstrap_servers,
             client_id,
             consumer_group,
+            kafka_consumer._client.cluster.cluster_id,
             self.args,
             self.kwargs,
         )
@@ -334,6 +334,7 @@ async def test_create_consumer_span(
             bootstrap_servers,
             client_id,
             consumer_group,
+            None,
             self.args,
             self.kwargs,
         )
@@ -358,6 +359,7 @@ async def test_create_consumer_span(
             partition=record.partition,
             key=str(record.key),
             offset=record.offset,
+            cluster_id=None,
         )
         consume_hook.assert_awaited_once_with(
             span, record, self.args, self.kwargs
@@ -365,7 +367,7 @@ async def test_create_consumer_span(
         detach.assert_called_once_with(attach.return_value)
 
     async def test_cluster_id_attribute_set_on_send_span(self) -> None:
-        """Cluster ID is added to producer span when cache is pre-populated."""
+        """Cluster ID is added to producer span when client metadata is available."""
         tracer = mock.MagicMock()
         span = mock.MagicMock()
         span.is_recording.return_value = True
@@ -376,36 +378,31 @@ async def test_cluster_id_attribute_set_on_send_span(self) -> None:
             return_value=False
         )
 
-        servers = "broker1:9092,broker2:9092"
-        cache_key = _normalize_bootstrap_servers(servers)
-        _cluster_id_cache[cache_key] = "test-cluster-uuid"
-        try:
-            producer = mock.MagicMock()
-            producer.client._bootstrap_servers = servers
-            producer.client._client_id = "test-client"
-            producer.client._wait_on_metadata = mock.AsyncMock()
-            producer._key_serializer = None
-            producer._value_serializer = None
-            producer._partition.return_value = 0
-
-            wrapped_send = _wrap_send(tracer, None)
-            await wrapped_send(
-                mock.AsyncMock(), producer, [self.topic_name], {}
-            )
+        producer = mock.MagicMock()
+        producer.client._bootstrap_servers = "broker1:9092,broker2:9092"
+        producer.client._client_id = "test-client"
+        producer.client._wait_on_metadata = mock.AsyncMock()
+        producer.client.cluster.cluster_id = "test-cluster-uuid"
+        producer._key_serializer = None
+        producer._value_serializer = None
+        producer._partition.return_value = 0
 
-            set_attribute_calls = {
-                call.args[0]: call.args[1]
-                for call in span.set_attribute.call_args_list
-            }
-            self.assertEqual(
-                set_attribute_calls.get(_MESSAGING_CLUSTER_ID),
-                "test-cluster-uuid",
-            )
-        finally:
-            _cluster_id_cache.pop(cache_key, None)
+        wrapped_send = _wrap_send(tracer, None)
+        await wrapped_send(
+            mock.AsyncMock(), producer, [self.topic_name], {}
+        )
+
+        set_attribute_calls = {
+            call.args[0]: call.args[1]
+            for call in span.set_attribute.call_args_list
+        }
+        self.assertEqual(
+            set_attribute_calls.get(_MESSAGING_CLUSTER_ID),
+            "test-cluster-uuid",
+        )
 
     async def test_cluster_id_attribute_absent_when_not_resolved(self) -> None:
-        """No cluster ID attribute is set when cache has no entry yet."""
+        """No cluster ID attribute is set when client metadata is not yet available."""
         tracer = mock.MagicMock()
         span = mock.MagicMock()
         span.is_recording.return_value = True
@@ -416,101 +413,57 @@ async def test_cluster_id_attribute_absent_when_not_resolved(self) -> None:
             return_value=False
         )
 
-        servers = "unknown-broker:9092"
-        cache_key = _normalize_bootstrap_servers(servers)
-        _cluster_id_cache.pop(cache_key, None)
-
-        with mock.patch(
-            "opentelemetry.instrumentation.aiokafka.utils._fetch_cluster_id_background"
-        ):
-            producer = mock.MagicMock()
-            producer.client._bootstrap_servers = servers
-            producer.client._client_id = "test-client"
-            producer.client._wait_on_metadata = mock.AsyncMock()
-            producer._key_serializer = None
-            producer._value_serializer = None
-            producer._partition.return_value = 0
-
-            wrapped_send = _wrap_send(tracer, None)
-            await wrapped_send(
-                mock.AsyncMock(), producer, [self.topic_name], {}
-            )
+        producer = mock.MagicMock()
+        producer.client._bootstrap_servers = "unknown-broker:9092"
+        producer.client._client_id = "test-client"
+        producer.client._wait_on_metadata = mock.AsyncMock()
+        producer.client.cluster.cluster_id = None
+        producer._key_serializer = None
+        producer._value_serializer = None
+        producer._partition.return_value = 0
+
+        wrapped_send = _wrap_send(tracer, None)
+        await wrapped_send(
+            mock.AsyncMock(), producer, [self.topic_name], {}
+        )
 
         attribute_keys = [
             call.args[0] for call in span.set_attribute.call_args_list
         ]
         self.assertNotIn(_MESSAGING_CLUSTER_ID, attribute_keys)
 
-    def test_normalize_bootstrap_servers_sorts_and_deduplicates_whitespace(
-        self,
-    ) -> None:
+    def test_extract_cluster_id_from_client_returns_cluster_id(self) -> None:
+        """Returns cluster ID from client.cluster.cluster_id when available."""
+        client = mock.MagicMock()
+        client.cluster.cluster_id = "abc-uuid-1234"
         self.assertEqual(
-            _normalize_bootstrap_servers("b:9092, a:9092"),
-            "a:9092,b:9092",
+            _extract_cluster_id_from_client(client), "abc-uuid-1234"
         )
-        self.assertEqual(
-            _normalize_bootstrap_servers(["b:9092", "a:9092"]),
-            "a:9092,b:9092",
-        )
-
-    def test_get_cluster_id_returns_none_for_sentinel(self) -> None:
-        servers = "sentinel-broker:9092"
-        cache_key = _normalize_bootstrap_servers(servers)
-        _cluster_id_cache[cache_key] = ""  # sentinel
-        try:
-            self.assertIsNone(_get_cluster_id(servers))
-        finally:
-            _cluster_id_cache.pop(cache_key, None)
-
-    def test_get_cluster_id_returns_resolved_value(self) -> None:
-        servers = "resolved-broker:9092"
-        cache_key = _normalize_bootstrap_servers(servers)
-        _cluster_id_cache[cache_key] = "abc-uuid"
-        try:
-            self.assertEqual(_get_cluster_id(servers), "abc-uuid")
-        finally:
-            _cluster_id_cache.pop(cache_key, None)
-
-    def test_fetch_cluster_id_background_sets_sentinel_and_spawns_thread(
+
+    def test_extract_cluster_id_from_client_returns_none_when_cluster_id_none(
         self,
     ) -> None:
-        servers = "fetch-test-broker:9092"
-        cache_key = _normalize_bootstrap_servers(servers)
-        _cluster_id_cache.pop(cache_key, None)
+        """Returns None when cluster_id is None (metadata not yet received)."""
+        client = mock.MagicMock()
+        client.cluster.cluster_id = None
+        self.assertIsNone(_extract_cluster_id_from_client(client))
 
-        with mock.patch("threading.Thread") as mock_thread:
-            mock_thread.return_value.start = mock.Mock()
-            _fetch_cluster_id_background(servers)
-            self.assertEqual(_cluster_id_cache.get(cache_key), "")
-            mock_thread.return_value.start.assert_called_once()
-
-        _cluster_id_cache.pop(cache_key, None)
-
-    def test_fetch_cluster_id_background_skips_if_sentinel_exists(
+    def test_extract_cluster_id_from_client_returns_none_when_no_cluster_attr(
         self,
     ) -> None:
-        servers = "sentinel-skip-broker:9092"
-        cache_key = _normalize_bootstrap_servers(servers)
-        _cluster_id_cache[cache_key] = ""  # pre-set sentinel
-        try:
-            with mock.patch("threading.Thread") as mock_thread:
-                _fetch_cluster_id_background(servers)
-                mock_thread.assert_not_called()
-        finally:
-            _cluster_id_cache.pop(cache_key, None)
-
-    def test_fetch_cluster_id_background_skips_if_already_resolved(
+        """Returns None when client has no cluster attribute."""
+        client = mock.MagicMock(spec=[])  # no attributes
+        self.assertIsNone(_extract_cluster_id_from_client(client))
+
+    def test_extract_cluster_id_from_client_returns_none_on_exception(
         self,
     ) -> None:
-        servers = "resolved-skip-broker:9092"
-        cache_key = _normalize_bootstrap_servers(servers)
-        _cluster_id_cache[cache_key] = "existing-uuid"
-        try:
-            with mock.patch("threading.Thread") as mock_thread:
-                _fetch_cluster_id_background(servers)
-                mock_thread.assert_not_called()
-        finally:
-            _cluster_id_cache.pop(cache_key, None)
+        """Returns None if attribute access raises unexpectedly."""
+        client = mock.MagicMock()
+        type(client).cluster = mock.PropertyMock(
+            side_effect=RuntimeError("boom")
+        )
+        self.assertIsNone(_extract_cluster_id_from_client(client))
 
     async def test_kafka_properties_extractor(self):
         aiokafka_instance_mock = mock.Mock()

From dd49f746bd184807eaf2339d3f5c8d049b3573fe Mon Sep 17 00:00:00 2001
From: Pulipelly Shashank Reddy <spulipelly@newrelic.com>
Date: Sat, 4 Jul 2026 16:48:48 +0530
Subject: [PATCH 09/14] instrumentation/aiokafka: gate messaging.cluster.id
 behind capture_experimental_span_attributes flag

Add a `capture_experimental_span_attributes: bool = False` kwarg to
`AIOKafkaInstrumentor().instrument()`, mirroring the Java instrumentation's
`captureExperimentalSpanAttributes` pattern.

`messaging.cluster.id` is in the experimental semconv namespace so it is
opt-in and off by default. The flag is threaded through `_wrap_send`,
`_wrap_getone`, and `_wrap_getmany` in utils.py. When False (default) the
cluster ID extraction is skipped entirely; when True the existing
lazy-populate-plus-post-send-retry logic runs as before.

Update unit tests: default assertions now expect cluster_id=None; the two
integration-style tests are updated to pass flag=True; a new test
`test_cluster_id_attribute_absent_by_default` verifies the flag-off path.

Assisted-by: Claude Sonnet 4.6
---
 .../instrumentation/aiokafka/__init__.py      | 13 ++++--
 .../instrumentation/aiokafka/utils.py         | 38 +++++++++++++----
 .../tests/test_utils.py                       | 41 ++++++++++++++++---
 3 files changed, 75 insertions(+), 17 deletions(-)

diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/__init__.py b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/__init__.py
index f694fb51be..716f8c48e0 100644
--- a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/__init__.py
+++ b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/__init__.py
@@ -126,6 +126,7 @@ class InstrumentKwargs(TypedDict, total=False):
         tracer_provider: trace.TracerProvider
         async_produce_hook: ProduceHookT
         async_consume_hook: ConsumeHookT
+        capture_experimental_span_attributes: bool
 
     class UninstrumentKwargs(TypedDict, total=False):
         pass
@@ -147,6 +148,8 @@ def _instrument(self, **kwargs: Unpack[InstrumentKwargs]):
                 ``tracer_provider``: a TracerProvider, defaults to global.
                 ``async_produce_hook``: a callable to be executed just before producing a message
                 ``async_consume_hook``: a callable to be executed just after consuming a message
+                ``capture_experimental_span_attributes``: if True, emit experimental
+                    ``messaging.cluster.id`` on producer and consumer spans. Defaults to False.
         """
         tracer_provider = kwargs.get("tracer_provider")
 
@@ -158,6 +161,10 @@ def _instrument(self, **kwargs: Unpack[InstrumentKwargs]):
         if not iscoroutinefunction(async_consume_hook):
             async_consume_hook = None
 
+        capture_experimental_span_attributes = kwargs.get(
+            "capture_experimental_span_attributes", False
+        )
+
         tracer = trace.get_tracer(
             __name__,
             __version__,
@@ -168,17 +175,17 @@ def _instrument(self, **kwargs: Unpack[InstrumentKwargs]):
         wrap_function_wrapper(
             aiokafka.AIOKafkaProducer,
             "send",
-            _wrap_send(tracer, async_produce_hook),
+            _wrap_send(tracer, async_produce_hook, capture_experimental_span_attributes),
         )
         wrap_function_wrapper(
             aiokafka.AIOKafkaConsumer,
             "getone",
-            _wrap_getone(tracer, async_consume_hook),
+            _wrap_getone(tracer, async_consume_hook, capture_experimental_span_attributes),
         )
         wrap_function_wrapper(
             aiokafka.AIOKafkaConsumer,
             "getmany",
-            _wrap_getmany(tracer, async_consume_hook),
+            _wrap_getmany(tracer, async_consume_hook, capture_experimental_span_attributes),
         )
 
     def _uninstrument(self, **kwargs: Unpack[UninstrumentKwargs]):
diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py
index 54f54f78be..1ca062708d 100644
--- a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py
+++ b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py
@@ -454,7 +454,9 @@ def _get_span_name(operation: str, topic: str):
 
 
 def _wrap_send(  # type: ignore[reportUnusedFunction]
-    tracer: Tracer, async_produce_hook: ProduceHookT | None
+    tracer: Tracer,
+    async_produce_hook: ProduceHookT | None,
+    capture_experimental_span_attributes: bool = False,
 ) -> Callable[..., Awaitable[asyncio.Future[RecordMetadata]]]:
     async def _traced_send(
         func: AIOKafkaSendProto,
@@ -473,6 +475,11 @@ async def _traced_send(
         client_id = _extract_client_id(instance.client)
         key = _deserialize_key(_extract_send_key(args, kwargs))
         partition = await _extract_send_partition(instance, args, kwargs)
+        cluster_id = (
+            _extract_cluster_id_from_client(instance.client)
+            if capture_experimental_span_attributes
+            else None
+        )
         span_name = _get_span_name("send", topic)
         with tracer.start_as_current_span(
             span_name, kind=trace.SpanKind.PRODUCER
@@ -484,7 +491,7 @@ async def _traced_send(
                 topic=topic,
                 partition=partition,
                 key=key,
-                cluster_id=_extract_cluster_id_from_client(instance.client),
+                cluster_id=cluster_id,
             )
             propagate.inject(
                 headers,
@@ -500,9 +507,10 @@ async def _traced_send(
             # After send() returns the broker has responded and cluster metadata
             # is populated.  Set the attribute now if it wasn't available at span
             # start (first send on a freshly-created producer).
-            cluster_id = _extract_cluster_id_from_client(instance.client)
-            if cluster_id is not None and span.is_recording():
-                span.set_attribute(_MESSAGING_CLUSTER_ID, cluster_id)
+            if capture_experimental_span_attributes:
+                cluster_id = _extract_cluster_id_from_client(instance.client)
+                if cluster_id is not None and span.is_recording():
+                    span.set_attribute(_MESSAGING_CLUSTER_ID, cluster_id)
             return result
 
     return _traced_send
@@ -551,7 +559,9 @@ async def _create_consumer_span(
 
 
 def _wrap_getone(  # type: ignore[reportUnusedFunction]
-    tracer: Tracer, async_consume_hook: ConsumeHookT | None
+    tracer: Tracer,
+    async_consume_hook: ConsumeHookT | None,
+    capture_experimental_span_attributes: bool = False,
 ) -> Callable[..., Awaitable[aiokafka.ConsumerRecord[object, object]]]:
     async def _traced_getone(
         func: AIOKafkaGetOneProto,
@@ -565,7 +575,11 @@ async def _traced_getone(
             bootstrap_servers = _extract_bootstrap_servers(instance._client)
             client_id = _extract_client_id(instance._client)
             consumer_group = _extract_consumer_group(instance)
-            cluster_id = _extract_cluster_id_from_client(instance._client)
+            cluster_id = (
+                _extract_cluster_id_from_client(instance._client)
+                if capture_experimental_span_attributes
+                else None
+            )
 
             extracted_context = propagate.extract(
                 record.headers, getter=_aiokafka_getter
@@ -588,7 +602,9 @@ async def _traced_getone(
 
 
 def _wrap_getmany(  # type: ignore[reportUnusedFunction]
-    tracer: Tracer, async_consume_hook: ConsumeHookT | None
+    tracer: Tracer,
+    async_consume_hook: ConsumeHookT | None,
+    capture_experimental_span_attributes: bool = False,
 ) -> Callable[
     ...,
     Awaitable[
@@ -612,7 +628,11 @@ async def _traced_getmany(
             bootstrap_servers = _extract_bootstrap_servers(instance._client)
             client_id = _extract_client_id(instance._client)
             consumer_group = _extract_consumer_group(instance)
-            cluster_id = _extract_cluster_id_from_client(instance._client)
+            cluster_id = (
+                _extract_cluster_id_from_client(instance._client)
+                if capture_experimental_span_attributes
+                else None
+            )
 
             span_name = _get_span_name(
                 "receive",
diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py
index 53a005f215..64019eca97 100644
--- a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py
+++ b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py
@@ -142,7 +142,7 @@ async def wrap_send_helper(
             topic=self.topic_name,
             partition=extract_send_partition.return_value,
             key=None,
-            cluster_id=kafka_producer.client.cluster.cluster_id,
+            cluster_id=None,
         )
 
         set_span_in_context.assert_called_once_with(span)
@@ -218,7 +218,7 @@ async def test_wrap_getone(
             bootstrap_servers,
             client_id,
             consumer_group,
-            kafka_consumer._client.cluster.cluster_id,
+            None,
             self.args,
             self.kwargs,
         )
@@ -300,7 +300,7 @@ async def test_wrap_getmany(
             bootstrap_servers,
             client_id,
             consumer_group,
-            kafka_consumer._client.cluster.cluster_id,
+            None,
             self.args,
             self.kwargs,
         )
@@ -387,7 +387,7 @@ async def test_cluster_id_attribute_set_on_send_span(self) -> None:
         producer._value_serializer = None
         producer._partition.return_value = 0
 
-        wrapped_send = _wrap_send(tracer, None)
+        wrapped_send = _wrap_send(tracer, None, capture_experimental_span_attributes=True)
         await wrapped_send(
             mock.AsyncMock(), producer, [self.topic_name], {}
         )
@@ -422,7 +422,38 @@ async def test_cluster_id_attribute_absent_when_not_resolved(self) -> None:
         producer._value_serializer = None
         producer._partition.return_value = 0
 
-        wrapped_send = _wrap_send(tracer, None)
+        wrapped_send = _wrap_send(tracer, None, capture_experimental_span_attributes=True)
+        await wrapped_send(
+            mock.AsyncMock(), producer, [self.topic_name], {}
+        )
+
+        attribute_keys = [
+            call.args[0] for call in span.set_attribute.call_args_list
+        ]
+        self.assertNotIn(_MESSAGING_CLUSTER_ID, attribute_keys)
+
+    async def test_cluster_id_attribute_absent_by_default(self) -> None:
+        """No cluster ID attribute is set when capture_experimental_span_attributes is False (default)."""
+        tracer = mock.MagicMock()
+        span = mock.MagicMock()
+        span.is_recording.return_value = True
+        tracer.start_as_current_span.return_value.__enter__ = mock.Mock(
+            return_value=span
+        )
+        tracer.start_as_current_span.return_value.__exit__ = mock.Mock(
+            return_value=False
+        )
+
+        producer = mock.MagicMock()
+        producer.client._bootstrap_servers = "broker1:9092"
+        producer.client._client_id = "test-client"
+        producer.client._wait_on_metadata = mock.AsyncMock()
+        producer.client.cluster.cluster_id = "some-cluster-uuid"
+        producer._key_serializer = None
+        producer._value_serializer = None
+        producer._partition.return_value = 0
+
+        wrapped_send = _wrap_send(tracer, None)  # default: flag=False
         await wrapped_send(
             mock.AsyncMock(), producer, [self.topic_name], {}
         )

From c54f4599cc49111a0982e0d8a2340b345904748b Mon Sep 17 00:00:00 2001
From: Pulipelly Shashank Reddy <spulipelly@newrelic.com>
Date: Sat, 4 Jul 2026 17:16:03 +0530
Subject: [PATCH 10/14] aiokafka: fix line length and add changelog fragment

- Break long _wrap_send/getone/getmany call arguments across lines
  to satisfy ruff's 79-character line limit
- Shorten two comment lines in _wrap_send that exceeded 79 chars
- Add .changelog/4727.added fragment for the messaging.cluster.id feature

Assisted-by: Claude Sonnet 4.6
---
 .../.changelog/.gitignore                      |  1 +
 .../.changelog/4727.added                      |  1 +
 .../instrumentation/aiokafka/__init__.py       | 18 +++++++++++++++---
 .../instrumentation/aiokafka/utils.py          |  5 ++---
 4 files changed, 19 insertions(+), 6 deletions(-)
 create mode 100644 instrumentation/opentelemetry-instrumentation-aiokafka/.changelog/.gitignore
 create mode 100644 instrumentation/opentelemetry-instrumentation-aiokafka/.changelog/4727.added

diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/.changelog/.gitignore b/instrumentation/opentelemetry-instrumentation-aiokafka/.changelog/.gitignore
new file mode 100644
index 0000000000..f935021a8f
--- /dev/null
+++ b/instrumentation/opentelemetry-instrumentation-aiokafka/.changelog/.gitignore
@@ -0,0 +1 @@
+!.gitignore
diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/.changelog/4727.added b/instrumentation/opentelemetry-instrumentation-aiokafka/.changelog/4727.added
new file mode 100644
index 0000000000..11db77dcc7
--- /dev/null
+++ b/instrumentation/opentelemetry-instrumentation-aiokafka/.changelog/4727.added
@@ -0,0 +1 @@
+`opentelemetry-instrumentation-aiokafka`: add `capture_experimental_span_attributes` option to gate `messaging.cluster.id` on producer and consumer spans
diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/__init__.py b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/__init__.py
index 716f8c48e0..8558a6a35d 100644
--- a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/__init__.py
+++ b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/__init__.py
@@ -175,17 +175,29 @@ def _instrument(self, **kwargs: Unpack[InstrumentKwargs]):
         wrap_function_wrapper(
             aiokafka.AIOKafkaProducer,
             "send",
-            _wrap_send(tracer, async_produce_hook, capture_experimental_span_attributes),
+            _wrap_send(
+                tracer,
+                async_produce_hook,
+                capture_experimental_span_attributes,
+            ),
         )
         wrap_function_wrapper(
             aiokafka.AIOKafkaConsumer,
             "getone",
-            _wrap_getone(tracer, async_consume_hook, capture_experimental_span_attributes),
+            _wrap_getone(
+                tracer,
+                async_consume_hook,
+                capture_experimental_span_attributes,
+            ),
         )
         wrap_function_wrapper(
             aiokafka.AIOKafkaConsumer,
             "getmany",
-            _wrap_getmany(tracer, async_consume_hook, capture_experimental_span_attributes),
+            _wrap_getmany(
+                tracer,
+                async_consume_hook,
+                capture_experimental_span_attributes,
+            ),
         )
 
     def _uninstrument(self, **kwargs: Unpack[UninstrumentKwargs]):
diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py
index 1ca062708d..09a0e4b2fc 100644
--- a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py
+++ b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py
@@ -504,9 +504,8 @@ async def _traced_send(
             except Exception as hook_exception:  # pylint: disable=W0703
                 _LOG.exception(hook_exception)
             result = await func(*args, **kwargs)
-            # After send() returns the broker has responded and cluster metadata
-            # is populated.  Set the attribute now if it wasn't available at span
-            # start (first send on a freshly-created producer).
+            # After send(), broker has responded and metadata is populated.
+            # Set cluster_id now if it wasn't available at span start.
             if capture_experimental_span_attributes:
                 cluster_id = _extract_cluster_id_from_client(instance.client)
                 if cluster_id is not None and span.is_recording():

From 36543955dea98b6fb04436b474cd987d0a3f018c Mon Sep 17 00:00:00 2001
From: Pulipelly Shashank Reddy <spulipelly@newrelic.com>
Date: Sat, 4 Jul 2026 17:25:25 +0530
Subject: [PATCH 11/14] kafka: apply ruff formatting fixes

Remove unused import and fix import sort order in aiokafka test_utils.py;
apply ruff-format to confluent-kafka and kafka-python utils/__init__.py

Assisted-by: Claude Sonnet 4.6
---
 .../tests/test_utils.py                       | 19 +++++------
 .../confluent_kafka/__init__.py               | 16 ++++++---
 .../instrumentation/confluent_kafka/utils.py  | 20 ++++++++---
 .../instrumentation/kafka/utils.py            | 34 ++++++++++++++-----
 4 files changed, 61 insertions(+), 28 deletions(-)

diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py
index 64019eca97..4806b778a2 100644
--- a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py
+++ b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py
@@ -7,11 +7,10 @@
 
 import aiokafka
 
-import opentelemetry.instrumentation.aiokafka.utils as aiokafka_utils
 from opentelemetry.instrumentation.aiokafka.utils import (
+    _MESSAGING_CLUSTER_ID,
     AIOKafkaContextGetter,
     AIOKafkaContextSetter,
-    _MESSAGING_CLUSTER_ID,
     _aiokafka_getter,
     _aiokafka_setter,
     _create_consumer_span,
@@ -387,10 +386,10 @@ async def test_cluster_id_attribute_set_on_send_span(self) -> None:
         producer._value_serializer = None
         producer._partition.return_value = 0
 
-        wrapped_send = _wrap_send(tracer, None, capture_experimental_span_attributes=True)
-        await wrapped_send(
-            mock.AsyncMock(), producer, [self.topic_name], {}
+        wrapped_send = _wrap_send(
+            tracer, None, capture_experimental_span_attributes=True
         )
+        await wrapped_send(mock.AsyncMock(), producer, [self.topic_name], {})
 
         set_attribute_calls = {
             call.args[0]: call.args[1]
@@ -422,10 +421,10 @@ async def test_cluster_id_attribute_absent_when_not_resolved(self) -> None:
         producer._value_serializer = None
         producer._partition.return_value = 0
 
-        wrapped_send = _wrap_send(tracer, None, capture_experimental_span_attributes=True)
-        await wrapped_send(
-            mock.AsyncMock(), producer, [self.topic_name], {}
+        wrapped_send = _wrap_send(
+            tracer, None, capture_experimental_span_attributes=True
         )
+        await wrapped_send(mock.AsyncMock(), producer, [self.topic_name], {})
 
         attribute_keys = [
             call.args[0] for call in span.set_attribute.call_args_list
@@ -454,9 +453,7 @@ async def test_cluster_id_attribute_absent_by_default(self) -> None:
         producer._partition.return_value = 0
 
         wrapped_send = _wrap_send(tracer, None)  # default: flag=False
-        await wrapped_send(
-            mock.AsyncMock(), producer, [self.topic_name], {}
-        )
+        await wrapped_send(mock.AsyncMock(), producer, [self.topic_name], {})
 
         attribute_keys = [
             call.args[0] for call in span.set_attribute.call_args_list
diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py
index c45ae6ce64..d75792553b 100644
--- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py
+++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py
@@ -149,7 +149,9 @@ def __init__(self, *args, **kwargs):
             self
         )
         if bootstrap_servers:
-            _fetch_cluster_id_background(bootstrap_servers, self.config, instance=self)
+            _fetch_cluster_id_background(
+                bootstrap_servers, self.config, instance=self
+            )
 
     # This method is deliberately implemented in order to allow wrapt to wrap this function
     def produce(self, topic, value=None, *args, **kwargs):  # pylint: disable=keyword-arg-before-vararg,useless-super-delegation
@@ -165,7 +167,9 @@ def __init__(self, *args, **kwargs):
             self
         )
         if bootstrap_servers:
-            _fetch_cluster_id_background(bootstrap_servers, self.config, instance=self)
+            _fetch_cluster_id_background(
+                bootstrap_servers, self.config, instance=self
+            )
 
     # This method is deliberately implemented in order to allow wrapt to wrap this function
     def poll(self, timeout=-1):  # pylint: disable=useless-super-delegation
@@ -192,7 +196,9 @@ def __init__(self, producer: Producer, tracer: Tracer):
             self
         )
         if bootstrap_servers:
-            _fetch_cluster_id_background(bootstrap_servers, self.config, instance=producer)
+            _fetch_cluster_id_background(
+                bootstrap_servers, self.config, instance=producer
+            )
 
     def flush(self, timeout=-1):
         return self._producer.flush(timeout)
@@ -228,7 +234,9 @@ def __init__(self, consumer: Consumer, tracer: Tracer):
             self
         )
         if bootstrap_servers:
-            _fetch_cluster_id_background(bootstrap_servers, self.config, instance=consumer)
+            _fetch_cluster_id_background(
+                bootstrap_servers, self.config, instance=consumer
+            )
 
     def close(self, *args, **kwargs):
         return ConfluentKafkaInstrumentor.wrap_close(
diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py
index 18fae3ace5..e7c56b6a31 100644
--- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py
+++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py
@@ -38,7 +38,11 @@
 
 def _get_real_instance(instance: Any) -> Any:
     """Unwrap Proxied* wrappers to get the underlying confluent-kafka Producer/Consumer."""
-    return getattr(instance, "_producer", None) or getattr(instance, "_consumer", None) or instance
+    return (
+        getattr(instance, "_producer", None)
+        or getattr(instance, "_consumer", None)
+        or instance
+    )
 
 
 def _bootstrap_cache_key(bootstrap_servers: Optional[str]) -> str:
@@ -65,7 +69,9 @@ def _fetch_cluster_id_background(
         # credentials (hot path) can still connect to SSL/SASL-protected clusters.
         if base_config is not None:
             _kafka_cluster_id_config_cache.setdefault(cache_key, base_config)
-        resolved_config = _kafka_cluster_id_config_cache.get(cache_key) or base_config
+        resolved_config = (
+            _kafka_cluster_id_config_cache.get(cache_key) or base_config
+        )
 
         existing = _kafka_cluster_id_cache.get(cache_key)
         if isinstance(existing, tuple):
@@ -84,7 +90,10 @@ def _run() -> None:
                 from confluent_kafka.admin import AdminClient  # noqa: PLC0415
 
                 admin = AdminClient(
-                    {**(resolved_config or {}), "bootstrap.servers": bootstrap_servers}
+                    {
+                        **(resolved_config or {}),
+                        "bootstrap.servers": bootstrap_servers,
+                    }
                 )
                 try:
                     cluster_metadata = admin.list_topics(timeout=10)
@@ -95,7 +104,10 @@ def _run() -> None:
             cluster_id = getattr(cluster_metadata, "cluster_id", None)
             if cluster_id:
                 with _kafka_cluster_id_lock:
-                    _kafka_cluster_id_cache[cache_key] = (cluster_id, time.monotonic())
+                    _kafka_cluster_id_cache[cache_key] = (
+                        cluster_id,
+                        time.monotonic(),
+                    )
             else:
                 with _kafka_cluster_id_lock:
                     _kafka_cluster_id_cache.pop(cache_key, None)
diff --git a/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/utils.py b/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/utils.py
index 4a83c2ce82..cbc95c0b99 100644
--- a/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/utils.py
+++ b/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/utils.py
@@ -19,13 +19,24 @@
 
 _MESSAGING_CLUSTER_ID = "messaging.cluster.id"
 
-_SECURITY_CONFIG_KEYS = frozenset({
-    "ssl_cafile", "ssl_certfile", "ssl_keyfile", "ssl_password",
-    "ssl_crlfile", "ssl_check_hostname", "ssl_context",
-    "security_protocol", "sasl_mechanism", "sasl_plain_username",
-    "sasl_plain_password", "sasl_kerberos_service_name",
-    "sasl_kerberos_domain_name", "sasl_oauth_token_provider",
-})
+_SECURITY_CONFIG_KEYS = frozenset(
+    {
+        "ssl_cafile",
+        "ssl_certfile",
+        "ssl_keyfile",
+        "ssl_password",
+        "ssl_crlfile",
+        "ssl_check_hostname",
+        "ssl_context",
+        "security_protocol",
+        "sasl_mechanism",
+        "sasl_plain_username",
+        "sasl_plain_password",
+        "sasl_kerberos_service_name",
+        "sasl_kerberos_domain_name",
+        "sasl_oauth_token_provider",
+    }
+)
 
 _CLUSTER_ID_TTL_SECONDS = 60 * 60
 
@@ -70,7 +81,10 @@ def _run() -> None:
             cluster_id = info.get("cluster_id")
             if cluster_id:
                 with _kafka_cluster_id_lock:
-                    _kafka_cluster_id_cache[cache_key] = (cluster_id, time.monotonic())
+                    _kafka_cluster_id_cache[cache_key] = (
+                        cluster_id,
+                        time.monotonic(),
+                    )
             else:
                 with _kafka_cluster_id_lock:
                     _kafka_cluster_id_cache.pop(cache_key, None)
@@ -301,7 +315,9 @@ def _traced_next(func, instance, args, kwargs):
             bootstrap_servers = (
                 KafkaPropertiesExtractor.extract_bootstrap_servers(instance)
             )
-            _fetch_cluster_id_background(bootstrap_servers, dict(instance.config))
+            _fetch_cluster_id_background(
+                bootstrap_servers, dict(instance.config)
+            )
             extracted_context = propagate.extract(
                 record.headers, getter=_kafka_getter
             )

From 90e6cf9ad5815090ac5bae7d1e7f2e5b2628e39f Mon Sep 17 00:00:00 2001
From: Pulipelly Shashank Reddy <spulipelly@newrelic.com>
Date: Sat, 4 Jul 2026 23:29:01 +0530
Subject: [PATCH 12/14] lint: fix pylint C0415/C0103 in confluent-kafka and
 kafka-python utils

- Add `# pylint: disable=import-outside-toplevel` alongside existing
  `# noqa: PLC0415` to suppress C0415 for lazy AdminClient imports
  (import is intentionally deferred to avoid hard dependency at module load)
- Rename thread variable `t` -> `thread` to satisfy C0103 naming pattern

Assisted-by: Claude Sonnet 4.6
---
 .../opentelemetry/instrumentation/confluent_kafka/utils.py  | 6 +++---
 .../src/opentelemetry/instrumentation/kafka/utils.py        | 6 +++---
 2 files changed, 6 insertions(+), 6 deletions(-)

diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py
index e7c56b6a31..96ccb5c680 100644
--- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py
+++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py
@@ -87,7 +87,7 @@ def _run() -> None:
             if instance is not None:
                 cluster_metadata = instance.list_topics(timeout=10)
             else:
-                from confluent_kafka.admin import AdminClient  # noqa: PLC0415
+                from confluent_kafka.admin import AdminClient  # noqa: PLC0415  # pylint: disable=import-outside-toplevel
 
                 admin = AdminClient(
                     {
@@ -115,11 +115,11 @@ def _run() -> None:
             with _kafka_cluster_id_lock:
                 _kafka_cluster_id_cache.pop(cache_key, None)
 
-    t = threading.Thread(
+    thread = threading.Thread(
         target=_run, daemon=True, name="otel-confluent-kafka-cluster-id"
     )
     try:
-        t.start()
+        thread.start()
     except Exception:  # pylint: disable=broad-except
         with _kafka_cluster_id_lock:
             _kafka_cluster_id_cache.pop(cache_key, None)
diff --git a/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/utils.py b/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/utils.py
index cbc95c0b99..653aaada96 100644
--- a/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/utils.py
+++ b/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/utils.py
@@ -67,7 +67,7 @@ def _fetch_cluster_id_background(bootstrap_servers, extra_config=None) -> None:
     def _run() -> None:
         admin = None
         try:
-            from kafka.admin import KafkaAdminClient  # noqa: PLC0415
+            from kafka.admin import KafkaAdminClient  # noqa: PLC0415  # pylint: disable=import-outside-toplevel
 
             security_kwargs = {
                 k: v
@@ -98,11 +98,11 @@ def _run() -> None:
                 except Exception:  # pylint: disable=broad-except
                     pass
 
-    t = threading.Thread(
+    thread = threading.Thread(
         target=_run, daemon=True, name="otel-kafka-cluster-id"
     )
     try:
-        t.start()
+        thread.start()
     except Exception:  # pylint: disable=broad-except
         with _kafka_cluster_id_lock:
             _kafka_cluster_id_cache.pop(cache_key, None)

From 5218073e06975f7ecc6574b4b7a9a8905ecb8c2d Mon Sep 17 00:00:00 2001
From: Pulipelly Shashank Reddy <spulipelly@newrelic.com>
Date: Sat, 4 Jul 2026 23:44:03 +0530
Subject: [PATCH 13/14] lint: move noqa comment to import line and wrap for
 isort

ruff-isort (I001) wraps long import lines to fit within the 79-char
line-length setting. Move the `# noqa: PLC0415` comment to the
opening `from ... import (` line so ruff honours the suppression
after wrapping.

Assisted-by: Claude Sonnet 4.6
---
 .../opentelemetry/instrumentation/confluent_kafka/utils.py    | 4 +++-
 .../src/opentelemetry/instrumentation/kafka/utils.py          | 4 +++-
 2 files changed, 6 insertions(+), 2 deletions(-)

diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py
index 96ccb5c680..ddadf8dd61 100644
--- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py
+++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py
@@ -87,7 +87,9 @@ def _run() -> None:
             if instance is not None:
                 cluster_metadata = instance.list_topics(timeout=10)
             else:
-                from confluent_kafka.admin import AdminClient  # noqa: PLC0415  # pylint: disable=import-outside-toplevel
+                from confluent_kafka.admin import (  # noqa: PLC0415
+                    AdminClient,
+                )
 
                 admin = AdminClient(
                     {
diff --git a/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/utils.py b/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/utils.py
index 653aaada96..eb5aee7ed5 100644
--- a/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/utils.py
+++ b/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/utils.py
@@ -67,7 +67,9 @@ def _fetch_cluster_id_background(bootstrap_servers, extra_config=None) -> None:
     def _run() -> None:
         admin = None
         try:
-            from kafka.admin import KafkaAdminClient  # noqa: PLC0415  # pylint: disable=import-outside-toplevel
+            from kafka.admin import (  # noqa: PLC0415
+                KafkaAdminClient,
+            )
 
             security_kwargs = {
                 k: v

From bda2608a2c233ae2946243f6cee4478887fbb176 Mon Sep 17 00:00:00 2001
From: Pulipelly Shashank Reddy <spulipelly@newrelic.com>
Date: Sat, 4 Jul 2026 23:56:58 +0530
Subject: [PATCH 14/14] fix(kafka,confluent-kafka): restore pylint disable on
 lazy admin imports

Re-add `# pylint: disable=import-outside-toplevel` to the lazy
`from ... import (` lines in both kafka and confluent-kafka utils.
The comment was accidentally dropped in the previous ruff-I001 fix
that moved the `# noqa: PLC0415` to the opening line of the
multi-line import form.

Assisted-by: Claude Sonnet 4.6
---
 .../src/opentelemetry/instrumentation/confluent_kafka/utils.py  | 2 +-
 .../src/opentelemetry/instrumentation/kafka/utils.py            | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py
index ddadf8dd61..a67268bc4b 100644
--- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py
+++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py
@@ -87,7 +87,7 @@ def _run() -> None:
             if instance is not None:
                 cluster_metadata = instance.list_topics(timeout=10)
             else:
-                from confluent_kafka.admin import (  # noqa: PLC0415
+                from confluent_kafka.admin import (  # noqa: PLC0415  # pylint: disable=import-outside-toplevel
                     AdminClient,
                 )
 
diff --git a/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/utils.py b/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/utils.py
index eb5aee7ed5..08b90b52dc 100644
--- a/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/utils.py
+++ b/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/utils.py
@@ -67,7 +67,7 @@ def _fetch_cluster_id_background(bootstrap_servers, extra_config=None) -> None:
     def _run() -> None:
         admin = None
         try:
-            from kafka.admin import (  # noqa: PLC0415
+            from kafka.admin import (  # noqa: PLC0415  # pylint: disable=import-outside-toplevel
                 KafkaAdminClient,
             )