From 15e62a31f4536b6b9668e0daccd6436f22553feb Mon Sep 17 00:00:00 2001 From: Pulipelly Shashank Reddy Date: Mon, 13 Jul 2026 23:56:53 +0530 Subject: [PATCH 01/20] feat(kafka): emit messaging.kafka.cluster.id unconditionally across kafka-python, confluent-kafka, and aiokafka Add always-on messaging.kafka.cluster.id span attribute to all three Python Kafka instrumentation libraries. Cluster ID is read lazily from the client instance (list_topics() for kafka-python/confluent-kafka, metadata() for aiokafka) with a 1-hour TTL cache per client. Removes the capture_experimental_span_attributes gate and promotes the attribute to default-on behavior matching the semconv stable promotion. Assisted-by: Claude Sonnet 4.6 --- .../.changelog/.gitignore | 1 + .../.changelog/4727.added | 1 + .../instrumentation/aiokafka/utils.py | 62 ++++++++- .../tests/test_utils.py | 105 +++++++++++++++ .../confluent_kafka/__init__.py | 33 +++++ .../instrumentation/confluent_kafka/utils.py | 122 +++++++++++++++++- .../instrumentation/kafka/__init__.py | 35 ++++- .../instrumentation/kafka/utils.py | 115 ++++++++++++++++- 8 files changed, 462 insertions(+), 12 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/utils.py b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py index ccfe157129..8b27ac78b3 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py @@ -79,6 +79,8 @@ async def __call__( _LOG = getLogger(__name__) +_MESSAGING_CLUSTER_ID = "messaging.kafka.cluster.id" + def _extract_bootstrap_servers( client: aiokafka.AIOKafkaClient, @@ -90,6 +92,24 @@ def _extract_client_id(client: aiokafka.AIOKafkaClient) -> str: return client._client_id +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( consumer: aiokafka.AIOKafkaConsumer, ) -> str | None: @@ -237,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, @@ -259,6 +280,9 @@ def _enrich_base_span( messaging_attributes.MESSAGING_KAFKA_MESSAGE_KEY, key ) + if cluster_id is not None: + span.set_attribute(_MESSAGING_CLUSTER_ID, cluster_id) + def _enrich_send_span( span: Span, @@ -268,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 @@ -279,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") @@ -298,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 @@ -309,6 +336,7 @@ def _enrich_getone_span( topic=topic, partition=partition, key=key, + cluster_id=cluster_id, ) if consumer_group is not None: @@ -344,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 @@ -357,6 +386,9 @@ def _enrich_getmany_poll_span( ) span.set_attribute(messaging_attributes.MESSAGING_CLIENT_ID, client_id) + 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 @@ -384,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 @@ -395,6 +428,7 @@ def _enrich_getmany_topic_span( topic=topic, partition=partition, key=None, + cluster_id=cluster_id, ) if consumer_group is not None: @@ -420,7 +454,8 @@ 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, ) -> Callable[..., Awaitable[asyncio.Future[RecordMetadata]]]: async def _traced_send( func: AIOKafkaSendProto, @@ -439,6 +474,7 @@ 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) span_name = _get_span_name("send", topic) with tracer.start_as_current_span( span_name, kind=trace.SpanKind.PRODUCER @@ -450,6 +486,7 @@ async def _traced_send( topic=topic, partition=partition, key=key, + cluster_id=cluster_id, ) propagate.inject( headers, @@ -461,8 +498,13 @@ 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(), broker has responded — refresh cluster ID in case + # metadata was not yet populated before the send started. + 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 @@ -475,6 +517,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: @@ -495,6 +538,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: @@ -508,7 +552,8 @@ 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, ) -> Callable[..., Awaitable[aiokafka.ConsumerRecord[object, object]]]: async def _traced_getone( func: AIOKafkaGetOneProto, @@ -522,6 +567,7 @@ 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) extracted_context = propagate.extract( record.headers, getter=_aiokafka_getter @@ -534,6 +580,7 @@ async def _traced_getone( bootstrap_servers, client_id, consumer_group, + cluster_id, args, kwargs, ) @@ -543,7 +590,8 @@ async def _traced_getone( def _wrap_getmany( # type: ignore[reportUnusedFunction] - tracer: Tracer, async_consume_hook: ConsumeHookT | None + tracer: Tracer, + async_consume_hook: ConsumeHookT | None, ) -> Callable[ ..., Awaitable[ @@ -567,6 +615,7 @@ 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) span_name = _get_span_name( "receive", @@ -581,6 +630,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(): @@ -596,6 +646,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: @@ -610,6 +661,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 0856d1389d..e363cd239a 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py @@ -8,11 +8,13 @@ import aiokafka from opentelemetry.instrumentation.aiokafka.utils import ( + _MESSAGING_CLUSTER_ID, AIOKafkaContextGetter, AIOKafkaContextSetter, _aiokafka_getter, _aiokafka_setter, _create_consumer_span, + _extract_cluster_id_from_client, _extract_send_partition, _get_span_name, _wrap_getmany, @@ -114,6 +116,7 @@ async def wrap_send_helper( produce_hook = mock.AsyncMock() original_send_callback = mock.AsyncMock() kafka_producer = mock.MagicMock() + kafka_producer.client.cluster.cluster_id = None expected_span_name = _get_span_name("send", self.topic_name) wrapped_send = _wrap_send(tracer, produce_hook) @@ -139,6 +142,7 @@ async def wrap_send_helper( topic=self.topic_name, partition=extract_send_partition.return_value, key=None, + cluster_id=None, ) set_span_in_context.assert_called_once_with(span) @@ -179,6 +183,7 @@ async def test_wrap_getone( consume_hook = mock.AsyncMock() original_getone_callback = mock.AsyncMock() kafka_consumer = mock.MagicMock() + kafka_consumer._client.cluster.cluster_id = None wrapped_getone = _wrap_getone(tracer, consume_hook) record = await wrapped_getone( @@ -214,6 +219,7 @@ async def test_wrap_getone( bootstrap_servers, client_id, consumer_group, + None, self.args, self.kwargs, ) @@ -259,6 +265,7 @@ async def test_wrap_getmany( } ) kafka_consumer = mock.MagicMock() + kafka_consumer._client.cluster.cluster_id = None _create_consumer_span.return_value = mock.MagicMock() wrapped_getmany = _wrap_getmany(tracer, consume_hook) @@ -295,6 +302,7 @@ async def test_wrap_getmany( bootstrap_servers, client_id, consumer_group, + None, self.args, self.kwargs, ) @@ -328,6 +336,7 @@ async def test_create_consumer_span( bootstrap_servers, client_id, consumer_group, + None, self.args, self.kwargs, ) @@ -352,12 +361,108 @@ 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 ) 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 client metadata is available.""" + 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,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 + + 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 client metadata is not yet available.""" + 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 = "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_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( + _extract_cluster_id_from_client(client), "abc-uuid-1234" + ) + + def test_extract_cluster_id_from_client_returns_none_when_cluster_id_none( + self, + ) -> 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)) + + def test_extract_cluster_id_from_client_returns_none_when_no_cluster_attr( + self, + ) -> None: + """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: + """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() 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..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 @@ -117,6 +117,8 @@ def instrument_consumer(consumer: Consumer, tracer_provider=None): ... _create_new_consume_span, _end_current_consume_span, _enrich_span, + _fetch_cluster_id_background, + _get_real_instance, _get_span_name, _kafka_setter, ) @@ -143,6 +145,13 @@ 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, 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 @@ -154,6 +163,13 @@ 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, 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 @@ -176,6 +192,13 @@ 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, instance=producer + ) def flush(self, timeout=-1): return self._producer.flush(timeout) @@ -207,6 +230,13 @@ 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, instance=consumer + ) def close(self, *args, **kwargs): return ConfluentKafkaInstrumentor.wrap_close( @@ -397,6 +427,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, @@ -425,6 +456,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) @@ -451,6 +483,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 27a188e7c7..55eeae6e4a 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,10 @@ # Copyright The OpenTelemetry Authors # SPDX-License-Identifier: Apache-2.0 +import threading +import time from logging import getLogger -from typing import List, Optional +from typing import Any, Dict, List, Optional from opentelemetry import context, propagate from opentelemetry.propagators import textmap @@ -24,6 +26,116 @@ _LOG = getLogger(__name__) +_MESSAGING_CLUSTER_ID = "messaging.kafka.cluster.id" + +_CLUSTER_ID_TTL_SECONDS = 60 * 60 + +_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 _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 "" + 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, + instance: Optional[Any] = None, +) -> None: + """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) + if not cache_key: + return + + with _kafka_cluster_id_lock: + 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: + return # still fresh; stale value stays until re-fetch succeeds + # TTL expired — leave stale tuple in cache so callers get the old value + # while the background refresh runs, then the refresh will overwrite it. + elif existing is not None: + return # "" sentinel — first fetch already in progress + else: + _kafka_cluster_id_cache[cache_key] = ( + "" # mark first fetch in-flight + ) + + def _run() -> None: + try: + if instance is not None: + cluster_metadata = instance.list_topics(timeout=10) + else: + from confluent_kafka.admin import ( # noqa: PLC0415 # pylint: disable=import-outside-toplevel + AdminClient, + ) + + 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(), + ) + 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) + + thread = threading.Thread( + target=_run, daemon=True, name="otel-confluent-kafka-cluster-id" + ) + try: + thread.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]: + if not bootstrap_servers: + return None + cache_key = _bootstrap_cache_key(bootstrap_servers) + val = _kafka_cluster_id_cache.get(cache_key) + return val[0] if isinstance(val, tuple) else None + class KafkaPropertiesExtractor: @staticmethod @@ -161,6 +273,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 @@ -178,11 +291,14 @@ 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) + _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) + # 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 5dd0df77e6..38ad1e2e15 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 @@ -90,7 +90,12 @@ def process_msg(message): from opentelemetry import trace from opentelemetry.instrumentation.instrumentor import BaseInstrumentor from opentelemetry.instrumentation.kafka.package import _instruments -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 @@ -123,6 +128,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) ) @@ -133,5 +164,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..2cd972b082 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,8 @@ # SPDX-License-Identifier: Apache-2.0 import json +import threading +import time from logging import getLogger from typing import Callable, Dict, List, Optional @@ -15,6 +17,108 @@ _LOG = getLogger(__name__) +_MESSAGING_CLUSTER_ID = "messaging.kafka.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", + } +) + +_CLUSTER_ID_TTL_SECONDS = 60 * 60 + +_kafka_cluster_id_cache: Dict[str, object] = {} +_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 isinstance(existing, tuple): + if time.monotonic() - existing[1] <= _CLUSTER_ID_TTL_SECONDS: + return # still fresh; stale value stays until re-fetch succeeds + # TTL expired — leave stale tuple in cache so callers get the old value + # while the background refresh runs, then the refresh will overwrite it. + elif existing is not None: + return # "" sentinel — first fetch already in progress + else: + _kafka_cluster_id_cache[cache_key] = ( + "" # mark first fetch in-flight + ) + + def _run() -> None: + admin = None + try: + from kafka.admin import ( # noqa: PLC0415 # pylint: disable=import-outside-toplevel + KafkaAdminClient, + ) + + 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, + time.monotonic(), + ) + 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 + + thread = threading.Thread( + target=_run, daemon=True, name="otel-kafka-cluster-id" + ) + try: + thread.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]: + cache_key = _bootstrap_cache_key(bootstrap_servers) + val = _kafka_cluster_id_cache.get(cache_key) + return val[0] if isinstance(val, tuple) else None + class KafkaPropertiesExtractor: @staticmethod @@ -135,6 +239,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): @@ -156,6 +263,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: @@ -170,8 +278,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 @@ -214,7 +321,9 @@ 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 193777e4c69d3708274441af88f1d3d48d11ac7d Mon Sep 17 00:00:00 2001 From: Pulipelly Shashank Reddy Date: Tue, 14 Jul 2026 12:35:29 +0530 Subject: [PATCH 02/20] feat(instrumentation/aiokafka): add messaging.kafka.cluster.id to producer/consumer spans aiokafka's ClusterMetadata receives cluster_id in every MetadataResponse but does not persist it as an attribute. Wrap cluster.update_metadata before start() to cache the cluster_id; read it from _extract_cluster_id_from_client at span creation time. Also refresh the attribute after send() in case the first metadata response arrived mid-send. Add _wrap_start_producer / _wrap_start_consumer wrappers and register them on AIOKafkaProducer.start / AIOKafkaConsumer.start in _instrument / _uninstrument. Tested E2E against PLAINTEXT, SASL/PLAIN, and SASL/SCRAM-SHA-256 listeners; messaging.kafka.cluster.id appears in all producer and consumer spans. Assisted-by: Claude Sonnet 4.6 --- .../instrumentation/aiokafka/__init__.py | 14 +++++ .../instrumentation/aiokafka/utils.py | 54 ++++++++++++++++++- .../tests/test_instrumentation.py | 12 +++++ .../tests/test_utils.py | 48 +++++++++++++++++ 4 files changed, 126 insertions(+), 2 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..cac438be0f 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/__init__.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/__init__.py @@ -109,6 +109,8 @@ async def produce(): _wrap_getmany, _wrap_getone, _wrap_send, + _wrap_start_consumer, + _wrap_start_producer, ) from opentelemetry.instrumentation.aiokafka.version import __version__ from opentelemetry.instrumentation.instrumentor import BaseInstrumentor @@ -165,6 +167,16 @@ def _instrument(self, **kwargs: Unpack[InstrumentKwargs]): schema_url=Schemas.V1_27_0.value, ) + wrap_function_wrapper( + aiokafka.AIOKafkaProducer, + "start", + _wrap_start_producer(), + ) + wrap_function_wrapper( + aiokafka.AIOKafkaConsumer, + "start", + _wrap_start_consumer(), + ) wrap_function_wrapper( aiokafka.AIOKafkaProducer, "send", @@ -182,6 +194,8 @@ def _instrument(self, **kwargs: Unpack[InstrumentKwargs]): ) def _uninstrument(self, **kwargs: Unpack[UninstrumentKwargs]): + unwrap(aiokafka.AIOKafkaProducer, "start") + unwrap(aiokafka.AIOKafkaConsumer, "start") unwrap(aiokafka.AIOKafkaProducer, "send") unwrap(aiokafka.AIOKafkaConsumer, "getone") unwrap(aiokafka.AIOKafkaConsumer, "getmany") 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 8b27ac78b3..5edd61b538 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py @@ -92,13 +92,33 @@ def _extract_client_id(client: aiokafka.AIOKafkaClient) -> str: return client._client_id +def _patch_cluster_id_capture(client: aiokafka.AIOKafkaClient) -> None: + """Wrap cluster.update_metadata once to populate cluster.cluster_id. + + aiokafka's ClusterMetadata receives cluster_id in every MetadataResponse + but does not store it as an attribute. This one-time patch intercepts + update_metadata so that _extract_cluster_id_from_client can read it. + """ + cluster = getattr(client, "cluster", None) + if cluster is None or getattr(cluster, "_otel_cluster_id_patched", False): + return + original_update = cluster.update_metadata + + def _patched_update(metadata: Any) -> None: + cluster_id = getattr(metadata, "cluster_id", None) + if cluster_id: + cluster.cluster_id = cluster_id + original_update(metadata) + + cluster.update_metadata = _patched_update + cluster._otel_cluster_id_patched = True + + 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: @@ -669,3 +689,33 @@ async def _traced_getmany( return records return _traced_getmany + + +def _wrap_start_producer() -> Callable[..., Awaitable[None]]: + """Wrap AIOKafkaProducer.start to install the cluster_id capture patch.""" + + async def _traced_start( + func: Callable[..., Awaitable[None]], + instance: aiokafka.AIOKafkaProducer, + args: tuple[Any, ...], + kwargs: dict[str, Any], + ) -> None: + _patch_cluster_id_capture(instance.client) + return await func(*args, **kwargs) + + return _traced_start + + +def _wrap_start_consumer() -> Callable[..., Awaitable[None]]: + """Wrap AIOKafkaConsumer.start to install the cluster_id capture patch.""" + + async def _traced_start( + func: Callable[..., Awaitable[None]], + instance: aiokafka.AIOKafkaConsumer, + args: tuple[Any, ...], + kwargs: dict[str, Any], + ) -> None: + _patch_cluster_id_capture(instance._client) + return await func(*args, **kwargs) + + return _traced_start diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_instrumentation.py b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_instrumentation.py index 35ec4bcb09..108bfdd266 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_instrumentation.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_instrumentation.py @@ -30,6 +30,12 @@ def test_instrument_api(self) -> None: instrumentation = AIOKafkaInstrumentor() instrumentation.instrument() + self.assertTrue( + isinstance(AIOKafkaProducer.start, BoundFunctionWrapper) + ) + self.assertTrue( + isinstance(AIOKafkaConsumer.start, BoundFunctionWrapper) + ) self.assertTrue( isinstance(AIOKafkaProducer.send, BoundFunctionWrapper) ) @@ -41,6 +47,12 @@ def test_instrument_api(self) -> None: ) instrumentation.uninstrument() + self.assertFalse( + isinstance(AIOKafkaProducer.start, BoundFunctionWrapper) + ) + self.assertFalse( + isinstance(AIOKafkaConsumer.start, BoundFunctionWrapper) + ) self.assertFalse( isinstance(AIOKafkaProducer.send, BoundFunctionWrapper) ) diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py index e363cd239a..4bc39a049e 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py @@ -17,6 +17,7 @@ _extract_cluster_id_from_client, _extract_send_partition, _get_span_name, + _patch_cluster_id_capture, _wrap_getmany, _wrap_getone, _wrap_send, @@ -430,6 +431,53 @@ async def test_cluster_id_attribute_absent_when_not_resolved(self) -> None: ] self.assertNotIn(_MESSAGING_CLUSTER_ID, attribute_keys) + def test_patch_cluster_id_capture_sets_cluster_id_from_metadata( + self, + ) -> None: + """_patch_cluster_id_capture intercepts update_metadata and sets cluster_id.""" + cluster = mock.MagicMock(spec=[]) + cluster.cluster_id = None + update_calls: list[object] = [] + + def original_update(metadata: object) -> None: + update_calls.append(metadata) + + cluster.update_metadata = original_update + client = mock.MagicMock() + client.cluster = cluster + + _patch_cluster_id_capture(client) + + metadata = mock.MagicMock() + metadata.cluster_id = "test-cluster-uuid" + cluster.update_metadata(metadata) + + self.assertEqual(cluster.cluster_id, "test-cluster-uuid") + self.assertEqual(update_calls, [metadata]) + + def test_patch_cluster_id_capture_is_idempotent(self) -> None: + """Calling _patch_cluster_id_capture twice does not double-wrap.""" + cluster = mock.MagicMock(spec=[]) + update_calls: list[object] = [] + cluster.update_metadata = lambda m: update_calls.append(m) + client = mock.MagicMock() + client.cluster = cluster + + _patch_cluster_id_capture(client) + _patch_cluster_id_capture(client) + + metadata = mock.MagicMock() + metadata.cluster_id = "id-1" + cluster.update_metadata(metadata) + + # original called exactly once despite two patch calls + self.assertEqual(len(update_calls), 1) + + def test_patch_cluster_id_capture_ignores_none_cluster(self) -> None: + """_patch_cluster_id_capture is a no-op when client has no cluster.""" + client = mock.MagicMock(spec=[]) # no attributes + _patch_cluster_id_capture(client) # must not raise + 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() From 1fe9ae1a3d2eec761e4c7341ffa8960c0672b92c Mon Sep 17 00:00:00 2001 From: Pulipelly Shashank Reddy Date: Tue, 14 Jul 2026 12:58:14 +0530 Subject: [PATCH 03/20] fix(instrumentation/aiokafka): move start wrappers inline to fix pyright/pylint Move _start_producer_wrapper and _start_consumer_wrapper from utils.py into __init__.py where they are used, eliminating the unnecessary factory pattern (no captured variables) and resolving: - pyright reportUnusedFunction: functions were flagged as unused because pyright checks within-file usage for module-level private functions - pylint W0108/R6301: remove unnecessary lambda, add @staticmethod to test_patch_cluster_id_capture_ignores_none_cluster Assisted-by: Claude Sonnet 4.6 --- .../instrumentation/aiokafka/__init__.py | 29 ++++++++++++++---- .../instrumentation/aiokafka/utils.py | 30 ------------------- .../tests/test_utils.py | 5 ++-- 3 files changed, 27 insertions(+), 37 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 cac438be0f..daa30f9e3f 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/__init__.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/__init__.py @@ -96,7 +96,7 @@ async def produce(): from __future__ import annotations from inspect import iscoroutinefunction -from typing import TYPE_CHECKING, Collection +from typing import TYPE_CHECKING, Any, Awaitable, Callable, Collection import aiokafka from wrapt import ( @@ -106,11 +106,10 @@ async def produce(): from opentelemetry import trace from opentelemetry.instrumentation.aiokafka.package import _instruments from opentelemetry.instrumentation.aiokafka.utils import ( + _patch_cluster_id_capture, _wrap_getmany, _wrap_getone, _wrap_send, - _wrap_start_consumer, - _wrap_start_producer, ) from opentelemetry.instrumentation.aiokafka.version import __version__ from opentelemetry.instrumentation.instrumentor import BaseInstrumentor @@ -133,6 +132,26 @@ class UninstrumentKwargs(TypedDict, total=False): pass +async def _start_producer_wrapper( + func: Callable[..., Awaitable[None]], + instance: aiokafka.AIOKafkaProducer, + args: tuple[Any, ...], + kwargs: dict[str, Any], +) -> None: + _patch_cluster_id_capture(instance.client) + await func(*args, **kwargs) + + +async def _start_consumer_wrapper( + func: Callable[..., Awaitable[None]], + instance: aiokafka.AIOKafkaConsumer, + args: tuple[Any, ...], + kwargs: dict[str, Any], +) -> None: + _patch_cluster_id_capture(instance._client) + await func(*args, **kwargs) + + class AIOKafkaInstrumentor(BaseInstrumentor): """An instrumentor for kafka module See `BaseInstrumentor` @@ -170,12 +189,12 @@ def _instrument(self, **kwargs: Unpack[InstrumentKwargs]): wrap_function_wrapper( aiokafka.AIOKafkaProducer, "start", - _wrap_start_producer(), + _start_producer_wrapper, ) wrap_function_wrapper( aiokafka.AIOKafkaConsumer, "start", - _wrap_start_consumer(), + _start_consumer_wrapper, ) wrap_function_wrapper( aiokafka.AIOKafkaProducer, 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 5edd61b538..d279b7575a 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py @@ -689,33 +689,3 @@ async def _traced_getmany( return records return _traced_getmany - - -def _wrap_start_producer() -> Callable[..., Awaitable[None]]: - """Wrap AIOKafkaProducer.start to install the cluster_id capture patch.""" - - async def _traced_start( - func: Callable[..., Awaitable[None]], - instance: aiokafka.AIOKafkaProducer, - args: tuple[Any, ...], - kwargs: dict[str, Any], - ) -> None: - _patch_cluster_id_capture(instance.client) - return await func(*args, **kwargs) - - return _traced_start - - -def _wrap_start_consumer() -> Callable[..., Awaitable[None]]: - """Wrap AIOKafkaConsumer.start to install the cluster_id capture patch.""" - - async def _traced_start( - func: Callable[..., Awaitable[None]], - instance: aiokafka.AIOKafkaConsumer, - args: tuple[Any, ...], - kwargs: dict[str, Any], - ) -> None: - _patch_cluster_id_capture(instance._client) - return await func(*args, **kwargs) - - return _traced_start diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py index 4bc39a049e..d95d1e7569 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py @@ -459,7 +459,7 @@ def test_patch_cluster_id_capture_is_idempotent(self) -> None: """Calling _patch_cluster_id_capture twice does not double-wrap.""" cluster = mock.MagicMock(spec=[]) update_calls: list[object] = [] - cluster.update_metadata = lambda m: update_calls.append(m) + cluster.update_metadata = update_calls.append client = mock.MagicMock() client.cluster = cluster @@ -473,7 +473,8 @@ def test_patch_cluster_id_capture_is_idempotent(self) -> None: # original called exactly once despite two patch calls self.assertEqual(len(update_calls), 1) - def test_patch_cluster_id_capture_ignores_none_cluster(self) -> None: + @staticmethod + def test_patch_cluster_id_capture_ignores_none_cluster() -> None: """_patch_cluster_id_capture is a no-op when client has no cluster.""" client = mock.MagicMock(spec=[]) # no attributes _patch_cluster_id_capture(client) # must not raise From e4c312b84190e38aebb26660d6c6db75e106f587 Mon Sep 17 00:00:00 2001 From: Pulipelly Shashank Reddy Date: Tue, 14 Jul 2026 13:04:52 +0530 Subject: [PATCH 04/20] fix(instrumentation/aiokafka): move _patch_cluster_id_capture to __init__ to satisfy pyright pyright reportUnusedFunction flags any module-level private function that is not accessed within the same file. _patch_cluster_id_capture was in utils.py but only called from __init__.py, triggering the error. Moving it to __init__.py where it is defined and called keeps all cross-file import graphs clean without requiring type: ignore annotations (prohibited by AGENTS.md). Update test_utils.py to import _patch_cluster_id_capture from __init__ instead. Assisted-by: Claude Sonnet 4.6 --- .../instrumentation/aiokafka/__init__.py | 17 +++++++++++++- .../instrumentation/aiokafka/utils.py | 22 ------------------- .../tests/test_utils.py | 2 +- 3 files changed, 17 insertions(+), 24 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 daa30f9e3f..c55a439ecb 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/__init__.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/__init__.py @@ -106,7 +106,6 @@ async def produce(): from opentelemetry import trace from opentelemetry.instrumentation.aiokafka.package import _instruments from opentelemetry.instrumentation.aiokafka.utils import ( - _patch_cluster_id_capture, _wrap_getmany, _wrap_getone, _wrap_send, @@ -132,6 +131,22 @@ class UninstrumentKwargs(TypedDict, total=False): pass +def _patch_cluster_id_capture(client: aiokafka.AIOKafkaClient) -> None: + cluster = getattr(client, "cluster", None) + if cluster is None or getattr(cluster, "_otel_cluster_id_patched", False): + return + original_update = cluster.update_metadata + + def _patched_update(metadata: Any) -> None: + cluster_id = getattr(metadata, "cluster_id", None) + if cluster_id: + cluster.cluster_id = cluster_id + original_update(metadata) + + cluster.update_metadata = _patched_update + cluster._otel_cluster_id_patched = True + + async def _start_producer_wrapper( func: Callable[..., Awaitable[None]], instance: aiokafka.AIOKafkaProducer, 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 d279b7575a..30042f30f5 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py @@ -92,28 +92,6 @@ def _extract_client_id(client: aiokafka.AIOKafkaClient) -> str: return client._client_id -def _patch_cluster_id_capture(client: aiokafka.AIOKafkaClient) -> None: - """Wrap cluster.update_metadata once to populate cluster.cluster_id. - - aiokafka's ClusterMetadata receives cluster_id in every MetadataResponse - but does not store it as an attribute. This one-time patch intercepts - update_metadata so that _extract_cluster_id_from_client can read it. - """ - cluster = getattr(client, "cluster", None) - if cluster is None or getattr(cluster, "_otel_cluster_id_patched", False): - return - original_update = cluster.update_metadata - - def _patched_update(metadata: Any) -> None: - cluster_id = getattr(metadata, "cluster_id", None) - if cluster_id: - cluster.cluster_id = cluster_id - original_update(metadata) - - cluster.update_metadata = _patched_update - cluster._otel_cluster_id_patched = True - - def _extract_cluster_id_from_client( client: aiokafka.AIOKafkaClient, ) -> str | None: diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py index d95d1e7569..95a2d27ae9 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py @@ -7,6 +7,7 @@ import aiokafka +from opentelemetry.instrumentation.aiokafka import _patch_cluster_id_capture from opentelemetry.instrumentation.aiokafka.utils import ( _MESSAGING_CLUSTER_ID, AIOKafkaContextGetter, @@ -17,7 +18,6 @@ _extract_cluster_id_from_client, _extract_send_partition, _get_span_name, - _patch_cluster_id_capture, _wrap_getmany, _wrap_getone, _wrap_send, From e1ceb1b969c8070c7191ab50a13b78af33736900 Mon Sep 17 00:00:00 2001 From: Pulipelly Shashank Reddy Date: Tue, 14 Jul 2026 18:29:36 +0530 Subject: [PATCH 05/20] fix(instrumentation/aiokafka): correct changelog entry for PR 4727 Assisted-by: Claude Sonnet 4.6 --- .../.changelog/4727.added | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/.changelog/4727.added b/instrumentation/opentelemetry-instrumentation-aiokafka/.changelog/4727.added index 11db77dcc7..f63468f4f4 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/.changelog/4727.added +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/.changelog/4727.added @@ -1 +1 @@ -`opentelemetry-instrumentation-aiokafka`: add `capture_experimental_span_attributes` option to gate `messaging.cluster.id` on producer and consumer spans +`opentelemetry-instrumentation-aiokafka`: emit `messaging.kafka.cluster.id` on producer and consumer spans From f26d6101351c6d459f0477cf9b7de1ca222eb6c1 Mon Sep 17 00:00:00 2001 From: Pulipelly Shashank Reddy Date: Wed, 15 Jul 2026 23:44:39 +0530 Subject: [PATCH 06/20] fix(instrumentation/kafka-python): source cluster id from client metadata, not a separate admin client Mirror the aiokafka instrumentation: read messaging.kafka.cluster.id from the client's own already-resolved metadata instead of opening a separate KafkaAdminClient in a background thread. Removes the hand-maintained security-config allowlist (which also omitted ssl_ciphers) and opens no extra broker connection. The id is captured from the MetadataResponse via update_metadata, so it works on kafka-python 2.0.x (which does not persist cluster_id on ClusterMetadata) as well as 2.1+. Assisted-by: Claude Opus 4.8 --- .../instrumentation/kafka/__init__.py | 19 +- .../instrumentation/kafka/utils.py | 163 ++++++------------ .../tests/test_utils.py | 67 ++++++- 3 files changed, 125 insertions(+), 124 deletions(-) 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 38ad1e2e15..26d57f5afd 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 @@ -91,8 +91,7 @@ def process_msg(message): from opentelemetry.instrumentation.instrumentor import BaseInstrumentor from opentelemetry.instrumentation.kafka.package import _instruments from opentelemetry.instrumentation.kafka.utils import ( - KafkaPropertiesExtractor, - _fetch_cluster_id_background, + _patch_cluster_id_capture, _wrap_next, _wrap_send, ) @@ -130,23 +129,11 @@ def _instrument(self, **kwargs): 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) - ) + _patch_cluster_id_capture(instance) 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) - ) + _patch_cluster_id_capture(instance) wrap_function_wrapper( kafka.KafkaProducer, "__init__", _wrap_producer_init 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 2cd972b082..59e0563dfc 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,8 +2,6 @@ # SPDX-License-Identifier: Apache-2.0 import json -import threading -import time from logging import getLogger from typing import Callable, Dict, List, Optional @@ -19,105 +17,47 @@ _MESSAGING_CLUSTER_ID = "messaging.kafka.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", - } -) - -_CLUSTER_ID_TTL_SECONDS = 60 * 60 - -_kafka_cluster_id_cache: Dict[str, object] = {} -_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 isinstance(existing, tuple): - if time.monotonic() - existing[1] <= _CLUSTER_ID_TTL_SECONDS: - return # still fresh; stale value stays until re-fetch succeeds - # TTL expired — leave stale tuple in cache so callers get the old value - # while the background refresh runs, then the refresh will overwrite it. - elif existing is not None: - return # "" sentinel — first fetch already in progress - else: - _kafka_cluster_id_cache[cache_key] = ( - "" # mark first fetch in-flight - ) - def _run() -> None: - admin = None - try: - from kafka.admin import ( # noqa: PLC0415 # pylint: disable=import-outside-toplevel - KafkaAdminClient, - ) +def _get_cluster_metadata(instance): + """Return the kafka-python ``ClusterMetadata`` for a producer or consumer. - 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, - time.monotonic(), - ) - 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 - - thread = threading.Thread( - target=_run, daemon=True, name="otel-kafka-cluster-id" - ) - try: - thread.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]: - cache_key = _bootstrap_cache_key(bootstrap_servers) - val = _kafka_cluster_id_cache.get(cache_key) - return val[0] if isinstance(val, tuple) else None + ``KafkaProducer`` exposes it as ``_metadata``; ``KafkaConsumer`` as + ``_client.cluster``. + """ + cluster = getattr(instance, "_metadata", None) + if cluster is not None: + return cluster + return getattr(getattr(instance, "_client", None), "cluster", None) + + +def _patch_cluster_id_capture(instance) -> None: + """Capture the cluster id from the client's own metadata responses. + + Reads from the client's already-resolved metadata; opens no extra broker + connection. kafka-python < 2.1 does not persist ``cluster_id`` on + ``ClusterMetadata``, but the ``MetadataResponse`` (v2+) passed to + ``update_metadata`` carries it, so wrap ``update_metadata`` to store it. + Guarded so each client's metadata object is patched at most once. + """ + cluster = _get_cluster_metadata(instance) + if cluster is None or getattr(cluster, "_otel_cluster_id_patched", False): + return + original_update = cluster.update_metadata + + def _patched_update(metadata): + result = original_update(metadata) + cluster_id = getattr(metadata, "cluster_id", None) + if cluster_id: + cluster.cluster_id = cluster_id + return result + + cluster.update_metadata = _patched_update + cluster._otel_cluster_id_patched = True + + +def _extract_cluster_id(instance) -> Optional[str]: + cluster_id = getattr(_get_cluster_metadata(instance), "cluster_id", None) + return cluster_id if cluster_id else None class KafkaPropertiesExtractor: @@ -230,7 +170,11 @@ def set(self, carrier: textmap.CarrierT, key: str, value: str) -> None: def _enrich_span( - span, bootstrap_servers: List[str], topic: str, partition: int + span, + bootstrap_servers: List[str], + topic: str, + partition: int, + cluster_id: Optional[str] = None, ): if span.is_recording(): span.set_attribute(SpanAttributes.MESSAGING_SYSTEM, "kafka") @@ -239,7 +183,6 @@ 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) @@ -262,12 +205,12 @@ def _traced_send(func, instance, args, kwargs): partition = KafkaPropertiesExtractor.extract_send_partition( instance, args, kwargs ) + cluster_id = _extract_cluster_id(instance) 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: - _enrich_span(span, bootstrap_servers, topic, partition) + _enrich_span(span, bootstrap_servers, topic, partition, cluster_id) propagate.inject( headers, context=trace.set_span_in_context(span), @@ -289,6 +232,7 @@ def _create_consumer_span( record, extracted_context, bootstrap_servers, + cluster_id, args, kwargs, ): @@ -300,7 +244,13 @@ def _create_consumer_span( ) as span: new_context = trace.set_span_in_context(span, extracted_context) token = context.attach(new_context) - _enrich_span(span, bootstrap_servers, record.topic, record.partition) + _enrich_span( + span, + bootstrap_servers, + record.topic, + record.partition, + cluster_id, + ) try: if callable(consume_hook): consume_hook(span, record, args, kwargs) @@ -321,9 +271,7 @@ def _traced_next(func, instance, args, kwargs): bootstrap_servers = ( KafkaPropertiesExtractor.extract_bootstrap_servers(instance) ) - _fetch_cluster_id_background( - bootstrap_servers, dict(instance.config) - ) + cluster_id = _extract_cluster_id(instance) extracted_context = propagate.extract( record.headers, getter=_kafka_getter ) @@ -333,6 +281,7 @@ def _traced_next(func, instance, args, kwargs): record, extracted_context, bootstrap_servers, + cluster_id, args, kwargs, ) diff --git a/instrumentation/opentelemetry-instrumentation-kafka-python/tests/test_utils.py b/instrumentation/opentelemetry-instrumentation-kafka-python/tests/test_utils.py index 3d5935a25f..f02625f1be 100644 --- a/instrumentation/opentelemetry-instrumentation-kafka-python/tests/test_utils.py +++ b/instrumentation/opentelemetry-instrumentation-kafka-python/tests/test_utils.py @@ -7,9 +7,11 @@ from opentelemetry.instrumentation.kafka.utils import ( KafkaPropertiesExtractor, _create_consumer_span, + _extract_cluster_id, _get_span_name, _kafka_getter, _kafka_setter, + _patch_cluster_id_capture, _wrap_next, _wrap_send, ) @@ -30,6 +32,9 @@ def setUp(self) -> None: @mock.patch( "opentelemetry.instrumentation.kafka.utils.KafkaPropertiesExtractor.extract_send_partition" ) + @mock.patch( + "opentelemetry.instrumentation.kafka.utils._extract_cluster_id" + ) @mock.patch("opentelemetry.instrumentation.kafka.utils._enrich_span") @mock.patch("opentelemetry.trace.set_span_in_context") @mock.patch("opentelemetry.propagate.inject") @@ -38,6 +43,7 @@ def test_wrap_send_with_topic_as_arg( inject: mock.MagicMock, set_span_in_context: mock.MagicMock, enrich_span: mock.MagicMock, + extract_cluster_id: mock.MagicMock, extract_send_partition: mock.MagicMock, extract_bootstrap_servers: mock.MagicMock, ) -> None: @@ -45,6 +51,7 @@ def test_wrap_send_with_topic_as_arg( inject, set_span_in_context, enrich_span, + extract_cluster_id, extract_send_partition, extract_bootstrap_servers, ) @@ -55,6 +62,9 @@ def test_wrap_send_with_topic_as_arg( @mock.patch( "opentelemetry.instrumentation.kafka.utils.KafkaPropertiesExtractor.extract_send_partition" ) + @mock.patch( + "opentelemetry.instrumentation.kafka.utils._extract_cluster_id" + ) @mock.patch("opentelemetry.instrumentation.kafka.utils._enrich_span") @mock.patch("opentelemetry.trace.set_span_in_context") @mock.patch("opentelemetry.propagate.inject") @@ -63,6 +73,7 @@ def test_wrap_send_with_topic_as_kwarg( inject: mock.MagicMock, set_span_in_context: mock.MagicMock, enrich_span: mock.MagicMock, + extract_cluster_id: mock.MagicMock, extract_send_partition: mock.MagicMock, extract_bootstrap_servers: mock.MagicMock, ) -> None: @@ -72,6 +83,7 @@ def test_wrap_send_with_topic_as_kwarg( inject, set_span_in_context, enrich_span, + extract_cluster_id, extract_send_partition, extract_bootstrap_servers, ) @@ -81,6 +93,7 @@ def wrap_send_helper( inject: mock.MagicMock, set_span_in_context: mock.MagicMock, enrich_span: mock.MagicMock, + extract_cluster_id: mock.MagicMock, extract_send_partition: mock.MagicMock, extract_bootstrap_servers: mock.MagicMock, ) -> None: @@ -109,6 +122,7 @@ def wrap_send_helper( extract_bootstrap_servers.return_value, self.topic_name, extract_send_partition.return_value, + extract_cluster_id.return_value, ) set_span_in_context.assert_called_once_with(span) @@ -124,6 +138,9 @@ def wrap_send_helper( ) self.assertEqual(retval, original_send_callback.return_value) + @mock.patch( + "opentelemetry.instrumentation.kafka.utils._extract_cluster_id" + ) @mock.patch("opentelemetry.propagate.extract") @mock.patch( "opentelemetry.instrumentation.kafka.utils._create_consumer_span" @@ -136,6 +153,7 @@ def test_wrap_next( extract_bootstrap_servers: mock.MagicMock, _create_consumer_span: mock.MagicMock, extract: mock.MagicMock, + extract_cluster_id: mock.MagicMock, ) -> None: tracer = mock.MagicMock() consume_hook = mock.MagicMock() @@ -164,6 +182,7 @@ def test_wrap_next( record, context, bootstrap_servers, + extract_cluster_id.return_value, self.args, self.kwargs, ) @@ -184,6 +203,7 @@ def test_create_consumer_span( bootstrap_servers = mock.MagicMock() extracted_context = mock.MagicMock() record = mock.MagicMock() + cluster_id = mock.MagicMock() _create_consumer_span( tracer, @@ -191,6 +211,7 @@ def test_create_consumer_span( record, extracted_context, bootstrap_servers, + cluster_id, self.args, self.kwargs, ) @@ -207,7 +228,11 @@ def test_create_consumer_span( attach.assert_called_once_with(set_span_in_context.return_value) enrich_span.assert_called_once_with( - span, bootstrap_servers, record.topic, record.partition + span, + bootstrap_servers, + record.topic, + record.partition, + cluster_id, ) consume_hook.assert_called_once_with( span, record, self.args, self.kwargs @@ -238,3 +263,43 @@ def test_kafka_properties_extractor( ) is None ) + + def test_extract_cluster_id_from_producer_metadata(self) -> None: + producer = mock.MagicMock() + producer._metadata.cluster_id = "test-cluster-id" + self.assertEqual(_extract_cluster_id(producer), "test-cluster-id") + + def test_extract_cluster_id_from_consumer_client(self) -> None: + consumer = mock.MagicMock(spec=["_client"]) + consumer._client.cluster.cluster_id = "test-cluster-id" + self.assertEqual(_extract_cluster_id(consumer), "test-cluster-id") + + def test_extract_cluster_id_absent_returns_none(self) -> None: + instance = mock.MagicMock(spec=[]) + self.assertIsNone(_extract_cluster_id(instance)) + + def test_patch_cluster_id_capture_captures_and_is_idempotent( + self, + ) -> None: + class FakeCluster: + def __init__(self) -> None: + self.update_calls = 0 + + def update_metadata(self, metadata) -> None: + self.update_calls += 1 + + class FakeMetadataResponse: + cluster_id = "test-cluster-id" + + cluster = FakeCluster() + instance = mock.MagicMock() + instance._metadata = cluster + + _patch_cluster_id_capture(instance) + # Second call must not double-wrap update_metadata. + _patch_cluster_id_capture(instance) + + cluster.update_metadata(FakeMetadataResponse()) + + self.assertEqual(cluster.update_calls, 1) + self.assertEqual(_extract_cluster_id(instance), "test-cluster-id") From d61b42821885dbd2a953e32bef113dca08f80f7e Mon Sep 17 00:00:00 2001 From: Pulipelly Shashank Reddy Date: Thu, 16 Jul 2026 00:00:01 +0530 Subject: [PATCH 07/20] fix: place PR 4727 changelog fragment in root .changelog/ for all three kafka packages aiokafka/confluent-kafka/kafka-python are coordinated packages, so their towncrier fragment belongs in the root .changelog/ directory (per CONTRIBUTING.md), not a package-level one. Move the fragment to .changelog/4727.added as a single entry with comma-separated package prefixes (matching the existing 4613.fixed fragment for the same packages), covering all three packages the PR touches. Remove the misplaced package-level .changelog/ directory and its self-referential .gitignore. Assisted-by: Claude Opus 4.8 --- .changelog/4727.added | 1 + .../opentelemetry-instrumentation-aiokafka/.changelog/.gitignore | 1 - .../opentelemetry-instrumentation-aiokafka/.changelog/4727.added | 1 - 3 files changed, 1 insertion(+), 2 deletions(-) create mode 100644 .changelog/4727.added delete mode 100644 instrumentation/opentelemetry-instrumentation-aiokafka/.changelog/.gitignore delete mode 100644 instrumentation/opentelemetry-instrumentation-aiokafka/.changelog/4727.added diff --git a/.changelog/4727.added b/.changelog/4727.added new file mode 100644 index 0000000000..d6c17de1ed --- /dev/null +++ b/.changelog/4727.added @@ -0,0 +1 @@ +`opentelemetry-instrumentation-aiokafka`, `opentelemetry-instrumentation-confluent-kafka`, `opentelemetry-instrumentation-kafka-python`: emit `messaging.kafka.cluster.id` on producer and consumer spans diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/.changelog/.gitignore b/instrumentation/opentelemetry-instrumentation-aiokafka/.changelog/.gitignore deleted file mode 100644 index f935021a8f..0000000000 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/.changelog/.gitignore +++ /dev/null @@ -1 +0,0 @@ -!.gitignore diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/.changelog/4727.added b/instrumentation/opentelemetry-instrumentation-aiokafka/.changelog/4727.added deleted file mode 100644 index f63468f4f4..0000000000 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/.changelog/4727.added +++ /dev/null @@ -1 +0,0 @@ -`opentelemetry-instrumentation-aiokafka`: emit `messaging.kafka.cluster.id` on producer and consumer spans From a419ba548788456381999a6a3a1e9b064d790982 Mon Sep 17 00:00:00 2001 From: Pulipelly Shashank Reddy Date: Thu, 16 Jul 2026 00:33:50 +0530 Subject: [PATCH 08/20] fix(instrumentation/kafka-python): keep test helper under pylint too-many-locals Adding the extract_cluster_id mock parameter pushed wrap_send_helper to 16 locals (pylint limit 15). Inline the single-use expected_span_name local to stay within the limit; no behavioral change. Assisted-by: Claude Opus 4.8 --- .../tests/test_utils.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/instrumentation/opentelemetry-instrumentation-kafka-python/tests/test_utils.py b/instrumentation/opentelemetry-instrumentation-kafka-python/tests/test_utils.py index f02625f1be..9b4841c846 100644 --- a/instrumentation/opentelemetry-instrumentation-kafka-python/tests/test_utils.py +++ b/instrumentation/opentelemetry-instrumentation-kafka-python/tests/test_utils.py @@ -101,7 +101,6 @@ def wrap_send_helper( produce_hook = mock.MagicMock() original_send_callback = mock.MagicMock() kafka_producer = mock.MagicMock() - expected_span_name = _get_span_name("send", self.topic_name) wrapped_send = _wrap_send(tracer, produce_hook) retval = wrapped_send( @@ -113,7 +112,7 @@ def wrap_send_helper( kafka_producer, self.args, self.kwargs ) tracer.start_as_current_span.assert_called_once_with( - expected_span_name, kind=SpanKind.PRODUCER + _get_span_name("send", self.topic_name), kind=SpanKind.PRODUCER ) span = tracer.start_as_current_span().__enter__.return_value From eefa7a5ee58a295297e9463391b83de2bbb20c7a Mon Sep 17 00:00:00 2001 From: Pulipelly Shashank Reddy Date: Thu, 16 Jul 2026 01:14:16 +0530 Subject: [PATCH 09/20] refactor(kafka): rename cluster-id constant to _MESSAGING_KAFKA_CLUSTER_ID + add semconv TODO The private constant was _MESSAGING_CLUSTER_ID, which dropped the 'kafka' segment. Rename it to _MESSAGING_KAFKA_CLUSTER_ID across the aiokafka, confluent-kafka and kafka-python instrumentations so it matches the attribute key (messaging.kafka.cluster.id) and the eventual generated semconv constant (messaging_attributes.MESSAGING_KAFKA_CLUSTER_ID). Add a TODO to switch to that constant once it is generated in opentelemetry-semantic-conventions (semconv spec PR #3819). Assisted-by: Claude Opus 4.8 --- .../opentelemetry/instrumentation/aiokafka/utils.py | 10 ++++++---- .../tests/test_utils.py | 6 +++--- .../instrumentation/confluent_kafka/utils.py | 6 ++++-- .../src/opentelemetry/instrumentation/kafka/utils.py | 6 ++++-- 4 files changed, 17 insertions(+), 11 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 30042f30f5..78ca1691c2 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py @@ -79,7 +79,9 @@ async def __call__( _LOG = getLogger(__name__) -_MESSAGING_CLUSTER_ID = "messaging.kafka.cluster.id" +# TODO(semconv #3819): once generated in opentelemetry-semantic-conventions, +# use messaging_attributes.MESSAGING_KAFKA_CLUSTER_ID instead of this literal. +_MESSAGING_KAFKA_CLUSTER_ID = "messaging.kafka.cluster.id" def _extract_bootstrap_servers( @@ -279,7 +281,7 @@ def _enrich_base_span( ) if cluster_id is not None: - span.set_attribute(_MESSAGING_CLUSTER_ID, cluster_id) + span.set_attribute(_MESSAGING_KAFKA_CLUSTER_ID, cluster_id) def _enrich_send_span( @@ -385,7 +387,7 @@ def _enrich_getmany_poll_span( span.set_attribute(messaging_attributes.MESSAGING_CLIENT_ID, client_id) if cluster_id is not None: - span.set_attribute(_MESSAGING_CLUSTER_ID, cluster_id) + span.set_attribute(_MESSAGING_KAFKA_CLUSTER_ID, cluster_id) if consumer_group is not None: span.set_attribute( @@ -501,7 +503,7 @@ async def _traced_send( # metadata was not yet populated before the send started. 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) + span.set_attribute(_MESSAGING_KAFKA_CLUSTER_ID, cluster_id) return result return _traced_send diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py index 95a2d27ae9..d3a637a309 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py @@ -9,7 +9,7 @@ from opentelemetry.instrumentation.aiokafka import _patch_cluster_id_capture from opentelemetry.instrumentation.aiokafka.utils import ( - _MESSAGING_CLUSTER_ID, + _MESSAGING_KAFKA_CLUSTER_ID, AIOKafkaContextGetter, AIOKafkaContextSetter, _aiokafka_getter, @@ -398,7 +398,7 @@ async def test_cluster_id_attribute_set_on_send_span(self) -> None: for call in span.set_attribute.call_args_list } self.assertEqual( - set_attribute_calls.get(_MESSAGING_CLUSTER_ID), + set_attribute_calls.get(_MESSAGING_KAFKA_CLUSTER_ID), "test-cluster-uuid", ) @@ -429,7 +429,7 @@ async def test_cluster_id_attribute_absent_when_not_resolved(self) -> None: attribute_keys = [ call.args[0] for call in span.set_attribute.call_args_list ] - self.assertNotIn(_MESSAGING_CLUSTER_ID, attribute_keys) + self.assertNotIn(_MESSAGING_KAFKA_CLUSTER_ID, attribute_keys) def test_patch_cluster_id_capture_sets_cluster_id_from_metadata( self, 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 55eeae6e4a..1cb6e59e14 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 @@ -26,7 +26,9 @@ _LOG = getLogger(__name__) -_MESSAGING_CLUSTER_ID = "messaging.kafka.cluster.id" +# TODO(semconv #3819): once generated in opentelemetry-semantic-conventions, +# use messaging_attributes.MESSAGING_KAFKA_CLUSTER_ID instead of this literal. +_MESSAGING_KAFKA_CLUSTER_ID = "messaging.kafka.cluster.id" _CLUSTER_ID_TTL_SECONDS = 60 * 60 @@ -297,7 +299,7 @@ def _enrich_span( _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) + span.set_attribute(_MESSAGING_KAFKA_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. 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 59e0563dfc..3ccc3a9979 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 @@ -15,7 +15,9 @@ _LOG = getLogger(__name__) -_MESSAGING_CLUSTER_ID = "messaging.kafka.cluster.id" +# TODO(semconv #3819): once generated in opentelemetry-semantic-conventions, +# use messaging_attributes.MESSAGING_KAFKA_CLUSTER_ID instead of this literal. +_MESSAGING_KAFKA_CLUSTER_ID = "messaging.kafka.cluster.id" def _get_cluster_metadata(instance): @@ -184,7 +186,7 @@ def _enrich_span( SpanAttributes.MESSAGING_URL, json.dumps(bootstrap_servers) ) if cluster_id: - span.set_attribute(_MESSAGING_CLUSTER_ID, cluster_id) + span.set_attribute(_MESSAGING_KAFKA_CLUSTER_ID, cluster_id) def _get_span_name(operation: str, topic: str): From f441b9b103ec33e80d67500364e154112dc53ba0 Mon Sep 17 00:00:00 2001 From: Pulipelly Shashank Reddy Date: Mon, 27 Jul 2026 14:47:07 +0530 Subject: [PATCH 10/20] confluent-kafka: replace background-thread cache with per-instance list_topics The previous implementation fetched cluster_id in a daemon thread using AdminClient when no live instance was available, and cached results in a module-level dict keyed by bootstrap.servers string. This had two problems: - The bootstrap.servers key is wrong when two distinct clusters share an address string; different Producer/Consumer instances cannot be told apart. - _fetch_cluster_id_background was called in _enrich_span (hot path) and in every __init__, causing lock contention and background threads on every instrumented object construction. Replace with a single non-blocking call to instance.list_topics(timeout=0), which reads librdkafka's internal metadata cache synchronously without any I/O or threads. Cache the result as instance._otel_cluster_id so subsequent spans are pure attribute reads. Remove all threading, time, and module-level cache globals. Add MockClusterMetadata and list_topics() to test helpers; add 5 tests. Assisted-by: Claude Sonnet 4.6 --- .../confluent_kafka/__init__.py | 29 ----- .../instrumentation/confluent_kafka/utils.py | 115 +++--------------- .../tests/test_instrumentation.py | 82 +++++++++++++ .../tests/utils.py | 15 +++ 4 files changed, 113 insertions(+), 128 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 d75792553b..e40c2a6554 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,7 +117,6 @@ def instrument_consumer(consumer: Consumer, tracer_provider=None): ... _create_new_consume_span, _end_current_consume_span, _enrich_span, - _fetch_cluster_id_background, _get_real_instance, _get_span_name, _kafka_setter, @@ -145,13 +144,6 @@ 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, 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 @@ -163,13 +155,6 @@ 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, 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,13 +177,6 @@ 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, instance=producer - ) def flush(self, timeout=-1): return self._producer.flush(timeout) @@ -230,13 +208,6 @@ 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, 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 1cb6e59e14..56bafebc91 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,10 +1,10 @@ # Copyright The OpenTelemetry Authors # SPDX-License-Identifier: Apache-2.0 -import threading -import time +from __future__ import annotations + from logging import getLogger -from typing import Any, Dict, List, Optional +from typing import Any, List, Optional from opentelemetry import context, propagate from opentelemetry.propagators import textmap @@ -30,13 +30,6 @@ # use messaging_attributes.MESSAGING_KAFKA_CLUSTER_ID instead of this literal. _MESSAGING_KAFKA_CLUSTER_ID = "messaging.kafka.cluster.id" -_CLUSTER_ID_TTL_SECONDS = 60 * 60 - -_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 _get_real_instance(instance: Any) -> Any: """Unwrap Proxied* wrappers to get the underlying confluent-kafka Producer/Consumer.""" @@ -47,96 +40,21 @@ def _get_real_instance(instance: Any) -> Any: ) -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, - instance: Optional[Any] = None, -) -> None: - """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) - if not cache_key: - return - - with _kafka_cluster_id_lock: - 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: - return # still fresh; stale value stays until re-fetch succeeds - # TTL expired — leave stale tuple in cache so callers get the old value - # while the background refresh runs, then the refresh will overwrite it. - elif existing is not None: - return # "" sentinel — first fetch already in progress - else: - _kafka_cluster_id_cache[cache_key] = ( - "" # mark first fetch in-flight - ) - - def _run() -> None: - try: - if instance is not None: - cluster_metadata = instance.list_topics(timeout=10) - else: - from confluent_kafka.admin import ( # noqa: PLC0415 # pylint: disable=import-outside-toplevel - AdminClient, - ) - - 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(), - ) - 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) - - thread = threading.Thread( - target=_run, daemon=True, name="otel-confluent-kafka-cluster-id" - ) +def _extract_cluster_id(instance: Any) -> Optional[str]: + """Read cluster_id from librdkafka's internal metadata cache. Non-blocking.""" + if instance is None: + return None + cached = getattr(instance, "_otel_cluster_id", None) + if cached: + return cached try: - thread.start() + cluster_metadata = instance.list_topics(timeout=0) + cluster_id = getattr(cluster_metadata, "cluster_id", None) + if cluster_id: + instance._otel_cluster_id = cluster_id + return cluster_id or None 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]: - if not bootstrap_servers: return None - cache_key = _bootstrap_cache_key(bootstrap_servers) - val = _kafka_cluster_id_cache.get(cache_key) - return val[0] if isinstance(val, tuple) else None class KafkaPropertiesExtractor: @@ -296,8 +214,7 @@ def _enrich_span( _set_bootstrap_servers_attributes(span, bootstrap_servers) - _fetch_cluster_id_background(bootstrap_servers, instance=instance) - cluster_id = _get_cluster_id(bootstrap_servers) + cluster_id = _extract_cluster_id(instance) if cluster_id: span.set_attribute(_MESSAGING_KAFKA_CLUSTER_ID, cluster_id) diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py index 8252d4ffb4..829f1cc578 100644 --- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py +++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py @@ -482,3 +482,85 @@ def test_consumer_sets_bootstrap_servers_attributes(self) -> None: ) self.assertEqual(process_span.attributes[SERVER_ADDRESS], "broker-1") self.assertEqual(process_span.attributes[SERVER_PORT], 9092) + + def test_cluster_id_set_on_producer_span(self) -> None: + instrumentation = ConfluentKafkaInstrumentor() + producer = MockedProducer( + [], + {"bootstrap.servers": "localhost:29092"}, + ) + producer._mock_cluster_id = "test-cluster-abc" + + producer = instrumentation.instrument_producer(producer) + producer.produce(topic="topic-1", key="k", value="v") + + span = self.memory_exporter.get_finished_spans()[0] + self.assertEqual( + span.attributes["messaging.kafka.cluster.id"], "test-cluster-abc" + ) + + def test_cluster_id_not_set_on_producer_span_when_unavailable(self) -> None: + instrumentation = ConfluentKafkaInstrumentor() + producer = MockedProducer( + [], + {"bootstrap.servers": "localhost:29092"}, + ) + # _mock_cluster_id defaults to None, so list_topics() returns None + + producer = instrumentation.instrument_producer(producer) + producer.produce(topic="topic-1", key="k", value="v") + + span = self.memory_exporter.get_finished_spans()[0] + self.assertNotIn("messaging.kafka.cluster.id", span.attributes) + + def test_cluster_id_set_on_consumer_poll_span(self) -> None: + instrumentation = ConfluentKafkaInstrumentor() + consumer = MockConsumer( + [MockedMessage("topic-1", 0, 0, [])], + { + "bootstrap.servers": "localhost:29092", + "group.id": "g", + "auto.offset.reset": "earliest", + }, + ) + consumer._mock_cluster_id = "test-cluster-xyz" + + self.memory_exporter.clear() + consumer = instrumentation.instrument_consumer(consumer) + consumer.poll() + consumer.poll() # end the in-flight process span + + process_span = next( + s + for s in self.memory_exporter.get_finished_spans() + if s.name == "topic-1 process" + ) + self.assertEqual( + process_span.attributes["messaging.kafka.cluster.id"], + "test-cluster-xyz", + ) + + def test_cluster_id_cached_on_instance(self) -> None: + from opentelemetry.instrumentation.confluent_kafka.utils import ( # noqa: PLC0415 + _extract_cluster_id, + ) + + producer = MockedProducer([], {"bootstrap.servers": "localhost:29092"}) + producer._mock_cluster_id = "cached-cluster-id" + + result = _extract_cluster_id(producer) + self.assertEqual(result, "cached-cluster-id") + # After first fetch, cached on instance — list_topics no longer needed + self.assertEqual(producer._otel_cluster_id, "cached-cluster-id") + + # Change mock to verify cache is read, not list_topics + producer._mock_cluster_id = "different-id" + result2 = _extract_cluster_id(producer) + self.assertEqual(result2, "cached-cluster-id") + + def test_extract_cluster_id_returns_none_for_none_instance(self) -> None: + from opentelemetry.instrumentation.confluent_kafka.utils import ( # noqa: PLC0415 + _extract_cluster_id, + ) + + self.assertIsNone(_extract_cluster_id(None)) diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/utils.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/utils.py index 62cbbe0ff7..7122771ca6 100644 --- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/utils.py +++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/utils.py @@ -6,10 +6,18 @@ from confluent_kafka import Consumer, Producer +class MockClusterMetadata: + def __init__(self, cluster_id: Optional[str] = None) -> None: + self.cluster_id = cluster_id + self.brokers: dict = {} + self.topics: dict = {} + + class MockConsumer(Consumer): def __init__(self, queue, config): self._queue = queue self.config = config + self._mock_cluster_id: Optional[str] = None super().__init__(config) def consume(self, num_messages=1, *args, **kwargs): # pylint: disable=keyword-arg-before-vararg @@ -22,6 +30,9 @@ def poll(self, timeout=None): return self._queue.pop(0) return None + def list_topics(self, topic=None, timeout=-1): + return MockClusterMetadata(cluster_id=self._mock_cluster_id) + class MockedMessage: def __init__( @@ -63,6 +74,7 @@ class MockedProducer(Producer): def __init__(self, queue, config): self._queue = queue self.config = config + self._mock_cluster_id: Optional[str] = None super().__init__(config) def produce(self, *args, **kwargs): # pylint: disable=keyword-arg-before-vararg @@ -82,3 +94,6 @@ def poll(self, *args, **kwargs): def flush(self, *args, **kwargs): return len(self._queue) + + def list_topics(self, topic=None, timeout=-1): + return MockClusterMetadata(cluster_id=self._mock_cluster_id) From 909a79e2802013cfa13e5ac09e79e71638775c61 Mon Sep 17 00:00:00 2001 From: Pulipelly Shashank Reddy Date: Mon, 27 Jul 2026 18:41:16 +0530 Subject: [PATCH 11/20] confluent-kafka: remove per-instance cluster_id cache from _extract_cluster_id MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The previous implementation stored the first successful cluster_id on the producer/consumer instance as `_otel_cluster_id` and skipped `list_topics()` on every subsequent span. This means a same-URL cluster migration (bootstrap URL unchanged but the underlying cluster replaced, e.g. blue/green) would permanently report the old cluster_id for the lifetime of the instance. `list_topics(timeout=0)` reads librdkafka's in-process metadata cache — it performs no I/O and costs a pointer dereference. Calling it on every span brings confluent-kafka in line with kafka-python, aiokafka, and the Java instrumentation, all of which read a live metadata object per span. Update the test to verify that a cluster_id change is visible immediately on the next span (migration-safe), rather than the old assertion that the stale cached value is returned. Assisted-by: Claude Sonnet 4.6 --- .../instrumentation/confluent_kafka/utils.py | 8 +----- .../tests/test_instrumentation.py | 26 ++++++++++--------- 2 files changed, 15 insertions(+), 19 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 56bafebc91..d5c7ba5f3f 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 @@ -44,15 +44,9 @@ def _extract_cluster_id(instance: Any) -> Optional[str]: """Read cluster_id from librdkafka's internal metadata cache. Non-blocking.""" if instance is None: return None - cached = getattr(instance, "_otel_cluster_id", None) - if cached: - return cached try: cluster_metadata = instance.list_topics(timeout=0) - cluster_id = getattr(cluster_metadata, "cluster_id", None) - if cluster_id: - instance._otel_cluster_id = cluster_id - return cluster_id or None + return getattr(cluster_metadata, "cluster_id", None) or None except Exception: # pylint: disable=broad-except return None diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py index 829f1cc578..c809004ffe 100644 --- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py +++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py @@ -32,7 +32,7 @@ from .utils import MockConsumer, MockedMessage, MockedProducer -class TestConfluentKafka(TestBase): +class TestConfluentKafka(TestBase): # pylint: disable=too-many-public-methods def test_instrument_api(self) -> None: from confluent_kafka import Consumer, Producer # noqa: PLC0415 @@ -499,7 +499,9 @@ def test_cluster_id_set_on_producer_span(self) -> None: span.attributes["messaging.kafka.cluster.id"], "test-cluster-abc" ) - def test_cluster_id_not_set_on_producer_span_when_unavailable(self) -> None: + def test_cluster_id_not_set_on_producer_span_when_unavailable( + self, + ) -> None: instrumentation = ConfluentKafkaInstrumentor() producer = MockedProducer( [], @@ -540,23 +542,23 @@ def test_cluster_id_set_on_consumer_poll_span(self) -> None: "test-cluster-xyz", ) - def test_cluster_id_cached_on_instance(self) -> None: + def test_cluster_id_reflects_current_value(self) -> None: from opentelemetry.instrumentation.confluent_kafka.utils import ( # noqa: PLC0415 _extract_cluster_id, ) producer = MockedProducer([], {"bootstrap.servers": "localhost:29092"}) - producer._mock_cluster_id = "cached-cluster-id" + producer._mock_cluster_id = "cluster-before-migration" - result = _extract_cluster_id(producer) - self.assertEqual(result, "cached-cluster-id") - # After first fetch, cached on instance — list_topics no longer needed - self.assertEqual(producer._otel_cluster_id, "cached-cluster-id") + self.assertEqual( + _extract_cluster_id(producer), "cluster-before-migration" + ) - # Change mock to verify cache is read, not list_topics - producer._mock_cluster_id = "different-id" - result2 = _extract_cluster_id(producer) - self.assertEqual(result2, "cached-cluster-id") + # Simulate cluster migration at same bootstrap URL — new cluster ID must be visible. + producer._mock_cluster_id = "cluster-after-migration" + self.assertEqual( + _extract_cluster_id(producer), "cluster-after-migration" + ) def test_extract_cluster_id_returns_none_for_none_instance(self) -> None: from opentelemetry.instrumentation.confluent_kafka.utils import ( # noqa: PLC0415 From 2632f87690dd82e74782b10ef65a0280ef81bf93 Mon Sep 17 00:00:00 2001 From: Pulipelly Shashank Reddy Date: Fri, 31 Jul 2026 19:07:21 +0530 Subject: [PATCH 12/20] confluent-kafka: skip list_topics() on consumers to avoid librdkafka UAF bug #4214 Calling list_topics() on a Consumer triggers a use-after-free in librdkafka (#4214). Producers are safe to call it; consumers must not. _extract_cluster_id now uses hasattr(instance, 'flush') to distinguish producers from consumers. Producers still call list_topics(timeout=0) (reads the in-process metadata cache, no I/O) and store the result in a new module-level dict _cluster_id_by_bootstrap keyed by bootstrap.servers. Consumers read that dict instead of calling list_topics(). Tests updated: test_cluster_id_set_on_consumer_poll_span now pre-populates _cluster_id_by_bootstrap as a producer would. Two new tests are added: test_cluster_id_not_set_on_consumer_span_when_bootstrap_cache_empty and test_consumer_does_not_call_list_topics. Assisted-by: Claude Sonnet 4.6 --- .../instrumentation/confluent_kafka/utils.py | 39 ++++++++--- .../tests/test_instrumentation.py | 67 ++++++++++++++++++- 2 files changed, 97 insertions(+), 9 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 d5c7ba5f3f..b5c245703e 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 @@ -40,15 +40,38 @@ def _get_real_instance(instance: Any) -> Any: ) -def _extract_cluster_id(instance: Any) -> Optional[str]: - """Read cluster_id from librdkafka's internal metadata cache. Non-blocking.""" +# Process-wide cache keyed by bootstrap.servers string; populated by producer spans so +# consumer spans can report cluster_id without calling list_topics() themselves. +_cluster_id_by_bootstrap: dict[str, str] = {} + + +def _extract_cluster_id( + instance: Any, bootstrap_servers: Optional[str] = None +) -> Optional[str]: + """Read cluster_id for span enrichment. + + Producers call list_topics(timeout=0) — reads librdkafka's in-process metadata + cache, no I/O — and the result is stored in _cluster_id_by_bootstrap. + Consumers never call list_topics(); they look up that cache by bootstrap + address. Calling list_topics() on a consumer is unsafe due to librdkafka + UAF bug #4214. + """ if instance is None: return None - try: - cluster_metadata = instance.list_topics(timeout=0) - return getattr(cluster_metadata, "cluster_id", None) or None - except Exception: # pylint: disable=broad-except - return None + if hasattr(instance, "flush"): + # Producer: list_topics() is safe here. + try: + cluster_metadata = instance.list_topics(timeout=0) + cluster_id = getattr(cluster_metadata, "cluster_id", None) or None + if cluster_id and bootstrap_servers: + _cluster_id_by_bootstrap[bootstrap_servers] = cluster_id + return cluster_id + except Exception: # pylint: disable=broad-except + return None + # Consumer: never call list_topics() — librdkafka UAF bug #4214. + if bootstrap_servers: + return _cluster_id_by_bootstrap.get(bootstrap_servers) + return None class KafkaPropertiesExtractor: @@ -208,7 +231,7 @@ def _enrich_span( _set_bootstrap_servers_attributes(span, bootstrap_servers) - cluster_id = _extract_cluster_id(instance) + cluster_id = _extract_cluster_id(instance, bootstrap_servers) if cluster_id: span.set_attribute(_MESSAGING_KAFKA_CLUSTER_ID, cluster_id) diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py index c809004ffe..eacc67b974 100644 --- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py +++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py @@ -516,6 +516,15 @@ def test_cluster_id_not_set_on_producer_span_when_unavailable( self.assertNotIn("messaging.kafka.cluster.id", span.attributes) def test_cluster_id_set_on_consumer_poll_span(self) -> None: + from opentelemetry.instrumentation.confluent_kafka.utils import ( # noqa: PLC0415 + _cluster_id_by_bootstrap, + ) + + # Consumers never call list_topics() (librdkafka UAF bug #4214); they read + # the bootstrap cache populated by producer spans on the same broker address. + _cluster_id_by_bootstrap["localhost:29092"] = "test-cluster-xyz" + self.addCleanup(_cluster_id_by_bootstrap.clear) + instrumentation = ConfluentKafkaInstrumentor() consumer = MockConsumer( [MockedMessage("topic-1", 0, 0, [])], @@ -525,7 +534,6 @@ def test_cluster_id_set_on_consumer_poll_span(self) -> None: "auto.offset.reset": "earliest", }, ) - consumer._mock_cluster_id = "test-cluster-xyz" self.memory_exporter.clear() consumer = instrumentation.instrument_consumer(consumer) @@ -542,6 +550,63 @@ def test_cluster_id_set_on_consumer_poll_span(self) -> None: "test-cluster-xyz", ) + def test_cluster_id_not_set_on_consumer_span_when_bootstrap_cache_empty( + self, + ) -> None: + from opentelemetry.instrumentation.confluent_kafka.utils import ( # noqa: PLC0415 + _cluster_id_by_bootstrap, + ) + + # No prior producer spans → cache is empty → consumer spans omit cluster_id. + _cluster_id_by_bootstrap.clear() + self.addCleanup(_cluster_id_by_bootstrap.clear) + + instrumentation = ConfluentKafkaInstrumentor() + consumer = MockConsumer( + [MockedMessage("topic-1", 0, 0, [])], + { + "bootstrap.servers": "localhost:29092", + "group.id": "g", + "auto.offset.reset": "earliest", + }, + ) + consumer._mock_cluster_id = "should-be-ignored" + + self.memory_exporter.clear() + consumer = instrumentation.instrument_consumer(consumer) + consumer.poll() + consumer.poll() + + process_span = next( + s + for s in self.memory_exporter.get_finished_spans() + if s.name == "topic-1 process" + ) + self.assertNotIn("messaging.kafka.cluster.id", process_span.attributes) + + def test_consumer_does_not_call_list_topics(self) -> None: + """list_topics() must never be called on consumers — librdkafka UAF bug #4214.""" + instrumentation = ConfluentKafkaInstrumentor() + consumer = MockConsumer( + [MockedMessage("topic-1", 0, 0, [])], + { + "bootstrap.servers": "localhost:29092", + "group.id": "g", + "auto.offset.reset": "earliest", + }, + ) + + def _fail_if_called(*args, **kwargs): + raise AssertionError( + "list_topics() called on a consumer — librdkafka UAF bug #4214" + ) + + consumer.list_topics = _fail_if_called + + consumer = instrumentation.instrument_consumer(consumer) + consumer.poll() + consumer.poll() + def test_cluster_id_reflects_current_value(self) -> None: from opentelemetry.instrumentation.confluent_kafka.utils import ( # noqa: PLC0415 _extract_cluster_id, From 905fe25d64e8660d36afd7639f3e50748497177d Mon Sep 17 00:00:00 2001 From: Pulipelly Shashank Reddy Date: Fri, 31 Jul 2026 23:11:23 +0530 Subject: [PATCH 13/20] test(confluent-kafka): add @staticmethod to test_consumer_does_not_call_list_topics Pylint R6301 (no-self-use): the method does not reference self. Assisted-by: Claude Sonnet 4.6 --- .../tests/test_instrumentation.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py index eacc67b974..f9be069a10 100644 --- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py +++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py @@ -584,7 +584,8 @@ def test_cluster_id_not_set_on_consumer_span_when_bootstrap_cache_empty( ) self.assertNotIn("messaging.kafka.cluster.id", process_span.attributes) - def test_consumer_does_not_call_list_topics(self) -> None: + @staticmethod + def test_consumer_does_not_call_list_topics() -> None: """list_topics() must never be called on consumers — librdkafka UAF bug #4214.""" instrumentation = ConfluentKafkaInstrumentor() consumer = MockConsumer( From c1d8ae5c8749c2ece82615053a27c838310aa3a0 Mon Sep 17 00:00:00 2001 From: Pulipelly Shashank Reddy Date: Fri, 31 Jul 2026 23:28:31 +0530 Subject: [PATCH 14/20] fix(confluent-kafka): move pylint disable inside class body to suppress R0904 Pylint 4.x evaluates too-many-public-methods at class teardown scope, so a disable comment on the class definition line is parsed in module scope and does not suppress the violation. Move the comment to the first line inside the class body where it takes effect for the whole class. Assisted-by: Claude Sonnet 4.6 --- .../tests/test_instrumentation.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py index f9be069a10..267ad46d6e 100644 --- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py +++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py @@ -32,7 +32,9 @@ from .utils import MockConsumer, MockedMessage, MockedProducer -class TestConfluentKafka(TestBase): # pylint: disable=too-many-public-methods +class TestConfluentKafka(TestBase): + # pylint: disable=too-many-public-methods + def test_instrument_api(self) -> None: from confluent_kafka import Consumer, Producer # noqa: PLC0415 From ab9849e4ed0891ce5c5c61f318624eae166ba9d1 Mon Sep 17 00:00:00 2001 From: Pulipelly Shashank Reddy Date: Sat, 1 Aug 2026 23:34:39 +0530 Subject: [PATCH 15/20] fix(aiokafka): replace update_metadata monkey-patch with MetadataRequest for cluster_id Use an explicit MetadataRequest_v5 wire-protocol call to fetch the Kafka cluster_id instead of monkey-patching aiokafka's internal cluster.update_metadata method, which relied on three volatile internal attribute names. The new approach: - _fetch_and_cache_cluster_id() is called once after producer/consumer start() - Sends a MetadataRequest_v5 directly to a random broker node - Caches the result on client._otel_cluster_id - Falls back to force_metadata_update() if no node is available yet - Applies a 5-minute backoff on failure to avoid hammering unreachable brokers - _extract_cluster_id_from_client() now reads client._otel_cluster_id Removes _patch_cluster_id_capture() and all its tests. Adds six new async tests for _fetch_and_cache_cluster_id covering: success, already-cached, failure backoff, no-node fallback, empty response, and send() exception. Assisted-by: Claude Sonnet 4.6 --- .../instrumentation/aiokafka/__init__.py | 21 +-- .../instrumentation/aiokafka/utils.py | 58 ++++++- .../tests/test_utils.py | 156 +++++++++++------- 3 files changed, 152 insertions(+), 83 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 c55a439ecb..5335ddc891 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/__init__.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/__init__.py @@ -106,6 +106,7 @@ async def produce(): from opentelemetry import trace from opentelemetry.instrumentation.aiokafka.package import _instruments from opentelemetry.instrumentation.aiokafka.utils import ( + _fetch_and_cache_cluster_id, _wrap_getmany, _wrap_getone, _wrap_send, @@ -131,30 +132,14 @@ class UninstrumentKwargs(TypedDict, total=False): pass -def _patch_cluster_id_capture(client: aiokafka.AIOKafkaClient) -> None: - cluster = getattr(client, "cluster", None) - if cluster is None or getattr(cluster, "_otel_cluster_id_patched", False): - return - original_update = cluster.update_metadata - - def _patched_update(metadata: Any) -> None: - cluster_id = getattr(metadata, "cluster_id", None) - if cluster_id: - cluster.cluster_id = cluster_id - original_update(metadata) - - cluster.update_metadata = _patched_update - cluster._otel_cluster_id_patched = True - - async def _start_producer_wrapper( func: Callable[..., Awaitable[None]], instance: aiokafka.AIOKafkaProducer, args: tuple[Any, ...], kwargs: dict[str, Any], ) -> None: - _patch_cluster_id_capture(instance.client) await func(*args, **kwargs) + await _fetch_and_cache_cluster_id(instance.client) async def _start_consumer_wrapper( @@ -163,8 +148,8 @@ async def _start_consumer_wrapper( args: tuple[Any, ...], kwargs: dict[str, Any], ) -> None: - _patch_cluster_id_capture(instance._client) await func(*args, **kwargs) + await _fetch_and_cache_cluster_id(instance._client) class AIOKafkaInstrumentor(BaseInstrumentor): 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 78ca1691c2..10f5b0ecc7 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 time from logging import getLogger from typing import ( TYPE_CHECKING, @@ -83,6 +84,15 @@ async def __call__( # use messaging_attributes.MESSAGING_KAFKA_CLUSTER_ID instead of this literal. _MESSAGING_KAFKA_CLUSTER_ID = "messaging.kafka.cluster.id" +_CLUSTER_ID_FAILURE_BACKOFF_SECS = 300 # 5 minutes + +try: + from aiokafka.protocol.metadata import ( # type: ignore[reportMissingImports] + MetadataRequest_v5 as _MetadataRequestV5, + ) +except ImportError: + _MetadataRequestV5 = None # type: ignore[assignment,misc] + def _extract_bootstrap_servers( client: aiokafka.AIOKafkaClient, @@ -97,17 +107,53 @@ def _extract_client_id(client: aiokafka.AIOKafkaClient) -> str: def _extract_cluster_id_from_client( client: aiokafka.AIOKafkaClient, ) -> str | None: - """Read cluster ID from the aiokafka client's cached cluster metadata. + """Return the cached cluster ID, or None if not yet resolved.""" + cluster_id: str | None = getattr(client, "_otel_cluster_id", None) + return cluster_id if cluster_id else None - Returns None if metadata has not been received yet. + +async def _fetch_and_cache_cluster_id( + client: aiokafka.AIOKafkaClient, +) -> None: + """Fetch cluster ID via a MetadataRequest and cache it on the client. + + Called once after start() completes. Subsequent calls return immediately + once the value is cached. Failed attempts are retried after + _CLUSTER_ID_FAILURE_BACKOFF_SECS to avoid hammering an unreachable broker. """ + if getattr(client, "_otel_cluster_id", None): + return + + failure_time: float | None = getattr( + client, "_otel_cluster_id_failure_time", None + ) + if ( + failure_time is not None + and time.monotonic() - failure_time < _CLUSTER_ID_FAILURE_BACKOFF_SECS + ): + return + + if _MetadataRequestV5 is None: + return + try: - cluster_id = getattr( - getattr(client, "cluster", None), "cluster_id", None + node_id = client.get_random_node() + if node_id is None: + await client.force_metadata_update() + node_id = client.get_random_node() + if node_id is None: + return + response = await client.send( + node_id, + _MetadataRequestV5(topics=[], allow_auto_topic_creation=False), ) - return cluster_id if cluster_id else None + cluster_id: str = getattr(response, "cluster_id", "") or "" + if cluster_id: + client._otel_cluster_id = cluster_id # type: ignore[attr-defined] + else: + client._otel_cluster_id_failure_time = time.monotonic() # type: ignore[attr-defined] except Exception: # pylint: disable=broad-except - return None + client._otel_cluster_id_failure_time = time.monotonic() # type: ignore[attr-defined] def _extract_consumer_group( diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py index d3a637a309..dbda88632a 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py @@ -3,11 +3,11 @@ # pylint: disable=unnecessary-dunder-call from __future__ import annotations +import time from unittest import IsolatedAsyncioTestCase, mock import aiokafka -from opentelemetry.instrumentation.aiokafka import _patch_cluster_id_capture from opentelemetry.instrumentation.aiokafka.utils import ( _MESSAGING_KAFKA_CLUSTER_ID, AIOKafkaContextGetter, @@ -17,6 +17,7 @@ _create_consumer_span, _extract_cluster_id_from_client, _extract_send_partition, + _fetch_and_cache_cluster_id, _get_span_name, _wrap_getmany, _wrap_getone, @@ -117,7 +118,7 @@ async def wrap_send_helper( produce_hook = mock.AsyncMock() original_send_callback = mock.AsyncMock() kafka_producer = mock.MagicMock() - kafka_producer.client.cluster.cluster_id = None + kafka_producer.client._otel_cluster_id = None expected_span_name = _get_span_name("send", self.topic_name) wrapped_send = _wrap_send(tracer, produce_hook) @@ -184,7 +185,7 @@ async def test_wrap_getone( consume_hook = mock.AsyncMock() original_getone_callback = mock.AsyncMock() kafka_consumer = mock.MagicMock() - kafka_consumer._client.cluster.cluster_id = None + kafka_consumer._client._otel_cluster_id = None wrapped_getone = _wrap_getone(tracer, consume_hook) record = await wrapped_getone( @@ -266,7 +267,7 @@ async def test_wrap_getmany( } ) kafka_consumer = mock.MagicMock() - kafka_consumer._client.cluster.cluster_id = None + kafka_consumer._client._otel_cluster_id = None _create_consumer_span.return_value = mock.MagicMock() wrapped_getmany = _wrap_getmany(tracer, consume_hook) @@ -385,7 +386,7 @@ async def test_cluster_id_attribute_set_on_send_span(self) -> None: 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.client._otel_cluster_id = "test-cluster-uuid" producer._key_serializer = None producer._value_serializer = None producer._partition.return_value = 0 @@ -418,7 +419,7 @@ async def test_cluster_id_attribute_absent_when_not_resolved(self) -> None: 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.client._otel_cluster_id = None producer._key_serializer = None producer._value_serializer = None producer._partition.return_value = 0 @@ -431,86 +432,123 @@ async def test_cluster_id_attribute_absent_when_not_resolved(self) -> None: ] self.assertNotIn(_MESSAGING_KAFKA_CLUSTER_ID, attribute_keys) - def test_patch_cluster_id_capture_sets_cluster_id_from_metadata( + def test_extract_cluster_id_from_client_returns_cluster_id(self) -> None: + """Returns cluster ID from _otel_cluster_id when available.""" + client = mock.MagicMock() + client._otel_cluster_id = "abc-uuid-1234" + self.assertEqual( + _extract_cluster_id_from_client(client), "abc-uuid-1234" + ) + + def test_extract_cluster_id_from_client_returns_none_when_not_set( self, ) -> None: - """_patch_cluster_id_capture intercepts update_metadata and sets cluster_id.""" - cluster = mock.MagicMock(spec=[]) - cluster.cluster_id = None - update_calls: list[object] = [] + """Returns None when _otel_cluster_id is not set on the client.""" + client = mock.MagicMock() + client._otel_cluster_id = None + self.assertIsNone(_extract_cluster_id_from_client(client)) - def original_update(metadata: object) -> None: - update_calls.append(metadata) + def test_extract_cluster_id_from_client_returns_none_when_no_attr( + self, + ) -> None: + """Returns None when client has no _otel_cluster_id attribute.""" + client = mock.MagicMock(spec=[]) # no attributes + self.assertIsNone(_extract_cluster_id_from_client(client)) - cluster.update_metadata = original_update + def test_extract_cluster_id_from_client_returns_none_on_empty_string( + self, + ) -> None: + """Returns None when _otel_cluster_id is an empty string.""" client = mock.MagicMock() - client.cluster = cluster + client._otel_cluster_id = "" + self.assertIsNone(_extract_cluster_id_from_client(client)) - _patch_cluster_id_capture(client) + async def test_fetch_and_cache_cluster_id_caches_on_success(self) -> None: + """Successfully fetched cluster_id is cached on the client.""" + client = mock.MagicMock() + client._otel_cluster_id = None + client._otel_cluster_id_failure_time = None + client.get_random_node.return_value = 0 + response = mock.MagicMock() + response.cluster_id = "abc-cluster-id" + client.send = mock.AsyncMock(return_value=response) - metadata = mock.MagicMock() - metadata.cluster_id = "test-cluster-uuid" - cluster.update_metadata(metadata) + await _fetch_and_cache_cluster_id(client) - self.assertEqual(cluster.cluster_id, "test-cluster-uuid") - self.assertEqual(update_calls, [metadata]) + self.assertEqual(client._otel_cluster_id, "abc-cluster-id") + client.send.assert_awaited_once() - def test_patch_cluster_id_capture_is_idempotent(self) -> None: - """Calling _patch_cluster_id_capture twice does not double-wrap.""" - cluster = mock.MagicMock(spec=[]) - update_calls: list[object] = [] - cluster.update_metadata = update_calls.append + async def test_fetch_and_cache_cluster_id_skips_if_already_cached( + self, + ) -> None: + """Does not send a request when cluster_id is already cached.""" client = mock.MagicMock() - client.cluster = cluster + client._otel_cluster_id = "already-cached" + client.send = mock.AsyncMock() - _patch_cluster_id_capture(client) - _patch_cluster_id_capture(client) + await _fetch_and_cache_cluster_id(client) - metadata = mock.MagicMock() - metadata.cluster_id = "id-1" - cluster.update_metadata(metadata) + client.send.assert_not_awaited() - # original called exactly once despite two patch calls - self.assertEqual(len(update_calls), 1) + async def test_fetch_and_cache_cluster_id_skips_during_backoff( + self, + ) -> None: + """Does not send a request during the failure backoff window.""" + client = mock.MagicMock() + client._otel_cluster_id = None + client._otel_cluster_id_failure_time = time.monotonic() + client.send = mock.AsyncMock() - @staticmethod - def test_patch_cluster_id_capture_ignores_none_cluster() -> None: - """_patch_cluster_id_capture is a no-op when client has no cluster.""" - client = mock.MagicMock(spec=[]) # no attributes - _patch_cluster_id_capture(client) # must not raise + await _fetch_and_cache_cluster_id(client) - 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( - _extract_cluster_id_from_client(client), "abc-uuid-1234" - ) + client.send.assert_not_awaited() - def test_extract_cluster_id_from_client_returns_none_when_cluster_id_none( + async def test_fetch_and_cache_cluster_id_force_update_when_no_node( self, ) -> None: - """Returns None when cluster_id is None (metadata not yet received).""" + """force_metadata_update is called when get_random_node returns None.""" client = mock.MagicMock() - client.cluster.cluster_id = None - self.assertIsNone(_extract_cluster_id_from_client(client)) + client._otel_cluster_id = None + client._otel_cluster_id_failure_time = None + client.get_random_node.return_value = None + client.force_metadata_update = mock.AsyncMock() + client.send = mock.AsyncMock() + + await _fetch_and_cache_cluster_id(client) - def test_extract_cluster_id_from_client_returns_none_when_no_cluster_attr( + client.force_metadata_update.assert_awaited_once() + client.send.assert_not_awaited() + + async def test_fetch_and_cache_cluster_id_empty_response_records_failure( self, ) -> None: - """Returns None when client has no cluster attribute.""" - client = mock.MagicMock(spec=[]) # no attributes + """Empty cluster_id in broker response records a failure time.""" + client = mock.MagicMock() + client._otel_cluster_id = None + client._otel_cluster_id_failure_time = None + client.get_random_node.return_value = 0 + response = mock.MagicMock() + response.cluster_id = "" + client.send = mock.AsyncMock(return_value=response) + + await _fetch_and_cache_cluster_id(client) + self.assertIsNone(_extract_cluster_id_from_client(client)) + self.assertIsNotNone(client._otel_cluster_id_failure_time) - def test_extract_cluster_id_from_client_returns_none_on_exception( + async def test_fetch_and_cache_cluster_id_exception_records_failure( self, ) -> None: - """Returns None if attribute access raises unexpectedly.""" + """send() exception records a failure time.""" client = mock.MagicMock() - type(client).cluster = mock.PropertyMock( - side_effect=RuntimeError("boom") - ) - self.assertIsNone(_extract_cluster_id_from_client(client)) + client._otel_cluster_id = None + client._otel_cluster_id_failure_time = None + client.get_random_node.return_value = 0 + client.send = mock.AsyncMock(side_effect=OSError("connection refused")) + + await _fetch_and_cache_cluster_id(client) + + self.assertIsNotNone(client._otel_cluster_id_failure_time) async def test_kafka_properties_extractor(self): aiokafka_instance_mock = mock.Mock() From a55cef3dd734748f508c7579d2b008106cd39cf7 Mon Sep 17 00:00:00 2001 From: Pulipelly Shashank Reddy Date: Sat, 1 Aug 2026 23:55:30 +0530 Subject: [PATCH 16/20] fix(aiokafka): resolve pyright and pylint errors in MetadataRequest cluster_id implementation Move _fetch_and_cache_cluster_id from utils.py to __init__.py where it is actually called, fixing pyright reportUnusedFunction. Use cast() to annotate untyped aiokafka get_random_node() and send() return values, fixing reportUnknownVariableType and reportUnknownMemberType. Remove unnecessary # type: ignore comments now flagged as reportUnnecessaryTypeIgnoreComment. Fix pylint R6301 no-self-use by using self.assertEqual() for await counts, and suppress R0904 too-many-public-methods in test file. Assisted-by: Claude Sonnet 4.6 --- .../instrumentation/aiokafka/__init__.py | 62 +++++++++++++++++-- .../instrumentation/aiokafka/utils.py | 55 ---------------- .../tests/test_utils.py | 12 ++-- 3 files changed, 63 insertions(+), 66 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 5335ddc891..4e21b4a1df 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/__init__.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/__init__.py @@ -95,18 +95,16 @@ async def produce(): from __future__ import annotations +import time from inspect import iscoroutinefunction -from typing import TYPE_CHECKING, Any, Awaitable, Callable, Collection +from typing import TYPE_CHECKING, Any, Awaitable, Callable, Collection, cast import aiokafka -from wrapt import ( - wrap_function_wrapper, # type: ignore[reportUnknownVariableType] -) +from wrapt import wrap_function_wrapper from opentelemetry import trace from opentelemetry.instrumentation.aiokafka.package import _instruments from opentelemetry.instrumentation.aiokafka.utils import ( - _fetch_and_cache_cluster_id, _wrap_getmany, _wrap_getone, _wrap_send, @@ -132,6 +130,60 @@ class UninstrumentKwargs(TypedDict, total=False): pass +_CLUSTER_ID_FAILURE_BACKOFF_SECS = 300 # 5 minutes + +try: + from aiokafka.protocol.metadata import ( + MetadataRequest_v5 as _MetadataRequestV5, + ) +except ImportError: + _MetadataRequestV5 = None + + +async def _fetch_and_cache_cluster_id( + client: aiokafka.AIOKafkaClient, +) -> None: + """Fetch cluster ID via a MetadataRequest and cache it on the client. + + Called once after start() completes. Subsequent calls return immediately + once the value is cached. Failed attempts are retried after + _CLUSTER_ID_FAILURE_BACKOFF_SECS to avoid hammering an unreachable broker. + """ + if getattr(client, "_otel_cluster_id", None): + return + + failure_time: float | None = getattr( + client, "_otel_cluster_id_failure_time", None + ) + if ( + failure_time is not None + and time.monotonic() - failure_time < _CLUSTER_ID_FAILURE_BACKOFF_SECS + ): + return + + if _MetadataRequestV5 is None: + return + + try: + node_id: int | None = cast(int | None, client.get_random_node()) + if node_id is None: + await client.force_metadata_update() + node_id = cast(int | None, client.get_random_node()) + if node_id is None: + return + response: object = await cast(Any, client).send( + node_id, + _MetadataRequestV5(topics=[], allow_auto_topic_creation=False), + ) + cluster_id: str = getattr(response, "cluster_id", "") or "" + if cluster_id: + client._otel_cluster_id = cluster_id # type: ignore[attr-defined] + else: + client._otel_cluster_id_failure_time = time.monotonic() # type: ignore[attr-defined] + except Exception: # pylint: disable=broad-except + client._otel_cluster_id_failure_time = time.monotonic() # type: ignore[attr-defined] + + async def _start_producer_wrapper( func: Callable[..., Awaitable[None]], instance: aiokafka.AIOKafkaProducer, 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 10f5b0ecc7..65d13ad6b0 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 time from logging import getLogger from typing import ( TYPE_CHECKING, @@ -84,16 +83,6 @@ async def __call__( # use messaging_attributes.MESSAGING_KAFKA_CLUSTER_ID instead of this literal. _MESSAGING_KAFKA_CLUSTER_ID = "messaging.kafka.cluster.id" -_CLUSTER_ID_FAILURE_BACKOFF_SECS = 300 # 5 minutes - -try: - from aiokafka.protocol.metadata import ( # type: ignore[reportMissingImports] - MetadataRequest_v5 as _MetadataRequestV5, - ) -except ImportError: - _MetadataRequestV5 = None # type: ignore[assignment,misc] - - def _extract_bootstrap_servers( client: aiokafka.AIOKafkaClient, ) -> str | list[str]: @@ -112,50 +101,6 @@ def _extract_cluster_id_from_client( return cluster_id if cluster_id else None -async def _fetch_and_cache_cluster_id( - client: aiokafka.AIOKafkaClient, -) -> None: - """Fetch cluster ID via a MetadataRequest and cache it on the client. - - Called once after start() completes. Subsequent calls return immediately - once the value is cached. Failed attempts are retried after - _CLUSTER_ID_FAILURE_BACKOFF_SECS to avoid hammering an unreachable broker. - """ - if getattr(client, "_otel_cluster_id", None): - return - - failure_time: float | None = getattr( - client, "_otel_cluster_id_failure_time", None - ) - if ( - failure_time is not None - and time.monotonic() - failure_time < _CLUSTER_ID_FAILURE_BACKOFF_SECS - ): - return - - if _MetadataRequestV5 is None: - return - - try: - node_id = client.get_random_node() - if node_id is None: - await client.force_metadata_update() - node_id = client.get_random_node() - if node_id is None: - return - response = await client.send( - node_id, - _MetadataRequestV5(topics=[], allow_auto_topic_creation=False), - ) - cluster_id: str = getattr(response, "cluster_id", "") or "" - if cluster_id: - client._otel_cluster_id = cluster_id # type: ignore[attr-defined] - else: - client._otel_cluster_id_failure_time = time.monotonic() # type: ignore[attr-defined] - except Exception: # pylint: disable=broad-except - client._otel_cluster_id_failure_time = time.monotonic() # type: ignore[attr-defined] - - def _extract_consumer_group( consumer: aiokafka.AIOKafkaConsumer, ) -> str | None: diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py index dbda88632a..ab71abf62e 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py @@ -1,6 +1,6 @@ # Copyright The OpenTelemetry Authors # SPDX-License-Identifier: Apache-2.0 -# pylint: disable=unnecessary-dunder-call +# pylint: disable=unnecessary-dunder-call,too-many-public-methods from __future__ import annotations import time @@ -8,6 +8,7 @@ import aiokafka +from opentelemetry.instrumentation.aiokafka import _fetch_and_cache_cluster_id from opentelemetry.instrumentation.aiokafka.utils import ( _MESSAGING_KAFKA_CLUSTER_ID, AIOKafkaContextGetter, @@ -17,7 +18,6 @@ _create_consumer_span, _extract_cluster_id_from_client, _extract_send_partition, - _fetch_and_cache_cluster_id, _get_span_name, _wrap_getmany, _wrap_getone, @@ -488,7 +488,7 @@ async def test_fetch_and_cache_cluster_id_skips_if_already_cached( await _fetch_and_cache_cluster_id(client) - client.send.assert_not_awaited() + self.assertEqual(client.send.await_count, 0) async def test_fetch_and_cache_cluster_id_skips_during_backoff( self, @@ -501,7 +501,7 @@ async def test_fetch_and_cache_cluster_id_skips_during_backoff( await _fetch_and_cache_cluster_id(client) - client.send.assert_not_awaited() + self.assertEqual(client.send.await_count, 0) async def test_fetch_and_cache_cluster_id_force_update_when_no_node( self, @@ -516,8 +516,8 @@ async def test_fetch_and_cache_cluster_id_force_update_when_no_node( await _fetch_and_cache_cluster_id(client) - client.force_metadata_update.assert_awaited_once() - client.send.assert_not_awaited() + self.assertEqual(client.force_metadata_update.await_count, 1) + self.assertEqual(client.send.await_count, 0) async def test_fetch_and_cache_cluster_id_empty_response_records_failure( self, From 0f747c476d52f7f1b5501a86fdd17ea4689e2809 Mon Sep 17 00:00:00 2001 From: Pulipelly Shashank Reddy Date: Sun, 2 Aug 2026 01:38:51 +0530 Subject: [PATCH 17/20] style(aiokafka): remove docstrings from instrumentation functions and tests Assisted-by: Claude Sonnet 4.6 --- .../opentelemetry/instrumentation/aiokafka/__init__.py | 6 ------ .../opentelemetry/instrumentation/aiokafka/utils.py | 1 - .../tests/test_utils.py | 10 ---------- 3 files changed, 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 4e21b4a1df..dbd22440ec 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/__init__.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/__init__.py @@ -143,12 +143,6 @@ class UninstrumentKwargs(TypedDict, total=False): async def _fetch_and_cache_cluster_id( client: aiokafka.AIOKafkaClient, ) -> None: - """Fetch cluster ID via a MetadataRequest and cache it on the client. - - Called once after start() completes. Subsequent calls return immediately - once the value is cached. Failed attempts are retried after - _CLUSTER_ID_FAILURE_BACKOFF_SECS to avoid hammering an unreachable broker. - """ if getattr(client, "_otel_cluster_id", None): return 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 65d13ad6b0..faa325a9d5 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py @@ -96,7 +96,6 @@ def _extract_client_id(client: aiokafka.AIOKafkaClient) -> str: def _extract_cluster_id_from_client( client: aiokafka.AIOKafkaClient, ) -> str | None: - """Return the cached cluster ID, or None if not yet resolved.""" cluster_id: str | None = getattr(client, "_otel_cluster_id", None) return cluster_id if cluster_id else None diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py index ab71abf62e..f87ad6ea04 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py @@ -433,7 +433,6 @@ async def test_cluster_id_attribute_absent_when_not_resolved(self) -> None: self.assertNotIn(_MESSAGING_KAFKA_CLUSTER_ID, attribute_keys) def test_extract_cluster_id_from_client_returns_cluster_id(self) -> None: - """Returns cluster ID from _otel_cluster_id when available.""" client = mock.MagicMock() client._otel_cluster_id = "abc-uuid-1234" self.assertEqual( @@ -443,7 +442,6 @@ def test_extract_cluster_id_from_client_returns_cluster_id(self) -> None: def test_extract_cluster_id_from_client_returns_none_when_not_set( self, ) -> None: - """Returns None when _otel_cluster_id is not set on the client.""" client = mock.MagicMock() client._otel_cluster_id = None self.assertIsNone(_extract_cluster_id_from_client(client)) @@ -451,20 +449,17 @@ def test_extract_cluster_id_from_client_returns_none_when_not_set( def test_extract_cluster_id_from_client_returns_none_when_no_attr( self, ) -> None: - """Returns None when client has no _otel_cluster_id 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_empty_string( self, ) -> None: - """Returns None when _otel_cluster_id is an empty string.""" client = mock.MagicMock() client._otel_cluster_id = "" self.assertIsNone(_extract_cluster_id_from_client(client)) async def test_fetch_and_cache_cluster_id_caches_on_success(self) -> None: - """Successfully fetched cluster_id is cached on the client.""" client = mock.MagicMock() client._otel_cluster_id = None client._otel_cluster_id_failure_time = None @@ -481,7 +476,6 @@ async def test_fetch_and_cache_cluster_id_caches_on_success(self) -> None: async def test_fetch_and_cache_cluster_id_skips_if_already_cached( self, ) -> None: - """Does not send a request when cluster_id is already cached.""" client = mock.MagicMock() client._otel_cluster_id = "already-cached" client.send = mock.AsyncMock() @@ -493,7 +487,6 @@ async def test_fetch_and_cache_cluster_id_skips_if_already_cached( async def test_fetch_and_cache_cluster_id_skips_during_backoff( self, ) -> None: - """Does not send a request during the failure backoff window.""" client = mock.MagicMock() client._otel_cluster_id = None client._otel_cluster_id_failure_time = time.monotonic() @@ -506,7 +499,6 @@ async def test_fetch_and_cache_cluster_id_skips_during_backoff( async def test_fetch_and_cache_cluster_id_force_update_when_no_node( self, ) -> None: - """force_metadata_update is called when get_random_node returns None.""" client = mock.MagicMock() client._otel_cluster_id = None client._otel_cluster_id_failure_time = None @@ -522,7 +514,6 @@ async def test_fetch_and_cache_cluster_id_force_update_when_no_node( async def test_fetch_and_cache_cluster_id_empty_response_records_failure( self, ) -> None: - """Empty cluster_id in broker response records a failure time.""" client = mock.MagicMock() client._otel_cluster_id = None client._otel_cluster_id_failure_time = None @@ -539,7 +530,6 @@ async def test_fetch_and_cache_cluster_id_empty_response_records_failure( async def test_fetch_and_cache_cluster_id_exception_records_failure( self, ) -> None: - """send() exception records a failure time.""" client = mock.MagicMock() client._otel_cluster_id = None client._otel_cluster_id_failure_time = None From 2fb269507693c847702348f24ff379d9a6c95400 Mon Sep 17 00:00:00 2001 From: Pulipelly Shashank Reddy Date: Sun, 2 Aug 2026 02:29:28 +0530 Subject: [PATCH 18/20] fix(confluent-kafka): allow consumers to call list_topics() as cache fallback Remove incorrect comment about a librdkafka bug that was unrelated to list_topics() safety. Consumers now call list_topics(timeout=0) when the bootstrap-servers cache is empty, matching producer behavior. Update tests to assert consumers receive cluster_id via list_topics() when no cached value exists. Assisted-by: Claude Sonnet 4.6 --- .../instrumentation/confluent_kafka/utils.py | 28 ++++++--------- .../tests/test_instrumentation.py | 34 +++---------------- 2 files changed, 15 insertions(+), 47 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 b5c245703e..a23f3eb2cc 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,7 +32,6 @@ 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) @@ -40,26 +39,15 @@ def _get_real_instance(instance: Any) -> Any: ) -# Process-wide cache keyed by bootstrap.servers string; populated by producer spans so -# consumer spans can report cluster_id without calling list_topics() themselves. _cluster_id_by_bootstrap: dict[str, str] = {} def _extract_cluster_id( instance: Any, bootstrap_servers: Optional[str] = None ) -> Optional[str]: - """Read cluster_id for span enrichment. - - Producers call list_topics(timeout=0) — reads librdkafka's in-process metadata - cache, no I/O — and the result is stored in _cluster_id_by_bootstrap. - Consumers never call list_topics(); they look up that cache by bootstrap - address. Calling list_topics() on a consumer is unsafe due to librdkafka - UAF bug #4214. - """ if instance is None: return None if hasattr(instance, "flush"): - # Producer: list_topics() is safe here. try: cluster_metadata = instance.list_topics(timeout=0) cluster_id = getattr(cluster_metadata, "cluster_id", None) or None @@ -68,10 +56,16 @@ def _extract_cluster_id( return cluster_id except Exception: # pylint: disable=broad-except return None - # Consumer: never call list_topics() — librdkafka UAF bug #4214. - if bootstrap_servers: - return _cluster_id_by_bootstrap.get(bootstrap_servers) - return None + if bootstrap_servers and bootstrap_servers in _cluster_id_by_bootstrap: + return _cluster_id_by_bootstrap[bootstrap_servers] + try: + cluster_metadata = instance.list_topics(timeout=0) + cluster_id = getattr(cluster_metadata, "cluster_id", None) or None + if cluster_id and bootstrap_servers: + _cluster_id_by_bootstrap[bootstrap_servers] = cluster_id + return cluster_id + except Exception: # pylint: disable=broad-except + return None class KafkaPropertiesExtractor: @@ -182,8 +176,6 @@ def _get_links_from_records(records): def _set_bootstrap_servers_attributes(span, bootstrap_servers): - """Populate server.address and server.port from a bootstrap.servers - string (e.g. ``host1:9092,host2:9092``).""" if not bootstrap_servers: return diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py index 267ad46d6e..625d12b3da 100644 --- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py +++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py @@ -522,8 +522,6 @@ def test_cluster_id_set_on_consumer_poll_span(self) -> None: _cluster_id_by_bootstrap, ) - # Consumers never call list_topics() (librdkafka UAF bug #4214); they read - # the bootstrap cache populated by producer spans on the same broker address. _cluster_id_by_bootstrap["localhost:29092"] = "test-cluster-xyz" self.addCleanup(_cluster_id_by_bootstrap.clear) @@ -552,14 +550,13 @@ def test_cluster_id_set_on_consumer_poll_span(self) -> None: "test-cluster-xyz", ) - def test_cluster_id_not_set_on_consumer_span_when_bootstrap_cache_empty( + def test_cluster_id_set_on_consumer_span_via_list_topics_when_cache_empty( self, ) -> None: from opentelemetry.instrumentation.confluent_kafka.utils import ( # noqa: PLC0415 _cluster_id_by_bootstrap, ) - # No prior producer spans → cache is empty → consumer spans omit cluster_id. _cluster_id_by_bootstrap.clear() self.addCleanup(_cluster_id_by_bootstrap.clear) @@ -572,7 +569,7 @@ def test_cluster_id_not_set_on_consumer_span_when_bootstrap_cache_empty( "auto.offset.reset": "earliest", }, ) - consumer._mock_cluster_id = "should-be-ignored" + consumer._mock_cluster_id = "cluster-from-list-topics" self.memory_exporter.clear() consumer = instrumentation.instrument_consumer(consumer) @@ -584,32 +581,11 @@ def test_cluster_id_not_set_on_consumer_span_when_bootstrap_cache_empty( for s in self.memory_exporter.get_finished_spans() if s.name == "topic-1 process" ) - self.assertNotIn("messaging.kafka.cluster.id", process_span.attributes) - - @staticmethod - def test_consumer_does_not_call_list_topics() -> None: - """list_topics() must never be called on consumers — librdkafka UAF bug #4214.""" - instrumentation = ConfluentKafkaInstrumentor() - consumer = MockConsumer( - [MockedMessage("topic-1", 0, 0, [])], - { - "bootstrap.servers": "localhost:29092", - "group.id": "g", - "auto.offset.reset": "earliest", - }, + self.assertEqual( + process_span.attributes["messaging.kafka.cluster.id"], + "cluster-from-list-topics", ) - def _fail_if_called(*args, **kwargs): - raise AssertionError( - "list_topics() called on a consumer — librdkafka UAF bug #4214" - ) - - consumer.list_topics = _fail_if_called - - consumer = instrumentation.instrument_consumer(consumer) - consumer.poll() - consumer.poll() - def test_cluster_id_reflects_current_value(self) -> None: from opentelemetry.instrumentation.confluent_kafka.utils import ( # noqa: PLC0415 _extract_cluster_id, From ccf030dc3058176d0a1177bedbd86515426c9a33 Mon Sep 17 00:00:00 2001 From: Pulipelly Shashank Reddy Date: Sun, 2 Aug 2026 02:29:36 +0530 Subject: [PATCH 19/20] fix(aiokafka): restore type: ignore for wrapt import, apply ruff format pyright 1.1.404 reports wrap_function_wrapper as partially unknown because wrapt uses complex callable types that pyright cannot fully resolve. The annotation is needed for CI typecheck to pass. Also apply ruff format to utils.py to fix pre-commit check failure. Assisted-by: Claude Sonnet 4.6 --- .../src/opentelemetry/instrumentation/aiokafka/__init__.py | 2 +- .../src/opentelemetry/instrumentation/aiokafka/utils.py | 1 + 2 files changed, 2 insertions(+), 1 deletion(-) 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 dbd22440ec..5f1433a995 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/__init__.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/__init__.py @@ -100,7 +100,7 @@ async def produce(): from typing import TYPE_CHECKING, Any, Awaitable, Callable, Collection, cast import aiokafka -from wrapt import wrap_function_wrapper +from wrapt import wrap_function_wrapper # type: ignore[reportUnknownVariableType] from opentelemetry import trace from opentelemetry.instrumentation.aiokafka.package import _instruments 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 faa325a9d5..00e893a8f9 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py @@ -83,6 +83,7 @@ async def __call__( # use messaging_attributes.MESSAGING_KAFKA_CLUSTER_ID instead of this literal. _MESSAGING_KAFKA_CLUSTER_ID = "messaging.kafka.cluster.id" + def _extract_bootstrap_servers( client: aiokafka.AIOKafkaClient, ) -> str | list[str]: From cccda089c09a763886d3dd8ad66f239f1ce0f8b9 Mon Sep 17 00:00:00 2001 From: Pulipelly Shashank Reddy Date: Sun, 2 Aug 2026 13:04:21 +0530 Subject: [PATCH 20/20] fix: sort wrapt import in aiokafka instrumentation Ruff I001 requires all imports to be sorted. The wrapt import needed parentheses to place the type-ignore comment on a separate continuation line while satisfying the sort order expected by the formatter. Assisted-by: Claude Sonnet 4.6 --- .../src/opentelemetry/instrumentation/aiokafka/__init__.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) 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 5f1433a995..4af71e90b5 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/__init__.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/__init__.py @@ -100,7 +100,9 @@ async def produce(): from typing import TYPE_CHECKING, Any, Awaitable, Callable, Collection, cast import aiokafka -from wrapt import wrap_function_wrapper # type: ignore[reportUnknownVariableType] +from wrapt import ( + wrap_function_wrapper, # type: ignore[reportUnknownVariableType] +) from opentelemetry import trace from opentelemetry.instrumentation.aiokafka.package import _instruments