feat(kafka): add messaging.kafka.cluster.id to producer/consumer spans by shashank-reddy-nr · Pull Request #4727 · open-telemetry/opentelemetry-python-contrib

shashank-reddy-nr · 2026-06-22T05:21:24Z

Description

Adds messaging.kafka.cluster.id (semconv) to Kafka producer and consumer spans across the aiokafka, kafka-python, and confluent-kafka instrumentations.

It is a Recommended semantic-convention attribute, so it is emitted by default (no opt-in flag) — consistent with the other Recommended Kafka attributes these instrumentations already produce. The id is read from each client's own already-resolved metadata, so no extra broker connection is opened:

aiokafka / kafka-python (pure-Python): read from the client's ClusterMetadata (cluster.cluster_id), captured from the broker MetadataResponse via update_metadata. This works on kafka-python 2.0.x (which does not persist cluster_id on ClusterMetadata) as well as 2.1+.
confluent-kafka (librdkafka): list_topics(timeout=0) is called on the producer/consumer instance on every span. With timeout=0 librdkafka returns immediately from its in-process metadata cache — no network I/O, no extra broker connection. This also means the attribute reflects cluster migration correctly (same bootstrap URL, new cluster): the cached value in librdkafka updates automatically when the client reconnects, and the next span picks it up.

The attribute may be briefly absent on the very first span before metadata is resolved; it self-heals on subsequent spans.

Type of change

New feature (non-breaking change which adds functionality)

How Has This Been Tested?

Unit tests added covering:

the attribute is set when the cluster id is available
the attribute is absent when broker metadata has not yet been received
(confluent-kafka) changing the cluster id is reflected immediately on the next span (migration-safe)

Does This PR Require a Core Repo Change?

No.

Checklist:

Followed the style guidelines of this project
Changelogs have been updated
Unit tests have been added
Documentation has been updated

linux-foundation-easycla · 2026-06-22T05:21:35Z

The committers listed above are authorized under a signed CLA.

✅ login: shashank-reddy-nr / name: Pulipelly Shashank Reddy (3760932)

shashank-reddy-nr · 2026-07-15T20:02:46Z

Hi @xrmx Can you please review this PR, whenever you have time?

…afka-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

…ducer/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

@staticmethod

…ght/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

…it__ 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

Assisted-by: Claude Sonnet 4.6

…data, 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

…ee 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

…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

…ER_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 open-telemetry#3819). Assisted-by: Claude Opus 4.8

…st_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

…luster_id 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

…UAF bug open-telemetry#4214 Calling list_topics() on a Consumer triggers a use-after-free in librdkafka (open-telemetry#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

…ll_list_topics Pylint R6301 (no-self-use): the method does not reference self. Assisted-by: Claude Sonnet 4.6

…ss 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

…est 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

…luster_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

… tests Assisted-by: Claude Sonnet 4.6

…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

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

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

emdneto

Thanks. Left some comments.

emdneto · 2026-08-03T23:39:30Z

    if operation:
        span.set_attribute(MESSAGING_OPERATION, operation.value)
-    else:
-        span.set_attribute(SpanAttributes.MESSAGING_TEMP_DESTINATION, True)


Why remove this?

emdneto · 2026-08-03T23:55:20Z

+        return None
+    if hasattr(instance, "flush"):
+        try:
+            cluster_metadata = instance.list_topics(timeout=0)


so for producers it will list_topics in every call? any overhead increase here for this hot path?

emdneto · 2026-08-04T00:07:57Z

+    kwargs: dict[str, Any],
+) -> None:
+    await func(*args, **kwargs)
+    await _fetch_and_cache_cluster_id(instance.client)


Something is odd with aiokafka tests after wrapping start.

emdneto · 2026-08-04T00:13:04Z


+# 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"


I understand that the attribute has already been added to the semantic conventions, but it hasn’t been released yet. Could we wait for the release before using it here? I’ll raise this at the next SIG meeting to get guidance on whether instrumentations should mix attributes from different semantic convention versions without the opt-in gate.

Yes, we can wait until the next semantic-conventions release. I recently learned that if an attribute hasn't been added to semantic-conventions yet, we should add an opt-in experimental attribute instead. Once the attribute is present and released in sem-conv, we can directly replace the current TODO with messaging_attributes.MESSAGING_KAFKA_CLUSTER_ID.

otelbot-python Bot added this to Python PR digest Jun 22, 2026

shashank-reddy-nr force-pushed the feature/kafka-cluster-id branch from 937e539 to 3760932 Compare June 22, 2026 13:05

shashank-reddy-nr marked this pull request as draft June 22, 2026 20:24

shashank-reddy-nr mentioned this pull request Jun 29, 2026

REQUEST: New membership for @shashank-reddy-nr open-telemetry/community#3556

Closed

6 tasks

shashank-reddy-nr changed the title ~~Feature/kafka cluster~~ instrumentation/aiokafka: add messaging.cluster.id to producer/consumer spans Jul 4, 2026

shashank-reddy-nr changed the title ~~instrumentation/aiokafka: add messaging.cluster.id to producer/consumer spans~~ instrumentation/aiokafka: add messaging.kafka.cluster.id to producer/consumer spans Jul 10, 2026

shashank-reddy-nr marked this pull request as ready for review July 10, 2026 14:27

shashank-reddy-nr changed the title ~~instrumentation/aiokafka: add messaging.kafka.cluster.id to producer/consumer spans~~ feat(instrumentation/aiokafka): add messaging.kafka.cluster.id to producer/consumer spans Jul 13, 2026

shashank-reddy-nr force-pushed the feature/kafka-cluster-id branch from 784ac81 to 185af6e Compare July 13, 2026 18:27

shashank-reddy-nr requested a review from a team as a code owner July 15, 2026 18:30

shashank-reddy-nr changed the title ~~feat(instrumentation/aiokafka): add messaging.kafka.cluster.id to producer/consumer spans~~ feat(kafka): add messaging.kafka.cluster.id to producer/consumer spans Jul 15, 2026

tammy-baylis-swi moved this to Ready for review in Python PR digest Jul 16, 2026

shashank-reddy-nr added 9 commits July 23, 2026 11:12

fix(instrumentation/aiokafka): correct changelog entry for PR 4727

e1ceb1b

Assisted-by: Claude Sonnet 4.6

shashank-reddy-nr force-pushed the feature/kafka-cluster-id branch from d53b17e to eefa7a5 Compare July 23, 2026 05:42

shashank-reddy-nr mentioned this pull request Jul 23, 2026

feat(stg): add OTel semconv messaging.kafka.cluster.id relationship rules newrelic/entity-definitions#3088

Merged

4 tasks

shashank-reddy-nr force-pushed the feature/kafka-cluster-id branch from c54c875 to 909a79e Compare July 27, 2026 13:16

shashank-reddy-nr added 10 commits July 31, 2026 19:07

test(confluent-kafka): add @staticmethod to test_consumer_does_not_ca…

905fe25

…ll_list_topics Pylint R6301 (no-self-use): the method does not reference self. Assisted-by: Claude Sonnet 4.6

style(aiokafka): remove docstrings from instrumentation functions and…

0f747c4

… tests Assisted-by: Claude Sonnet 4.6

fix: sort wrapt import in aiokafka instrumentation

cccda08

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

Merge branch 'main' into feature/kafka-cluster-id

c0fbf81

emdneto reviewed Aug 4, 2026

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

feat(kafka): add messaging.kafka.cluster.id to producer/consumer spans - #4727

feat(kafka): add messaging.kafka.cluster.id to producer/consumer spans#4727
shashank-reddy-nr wants to merge 21 commits into
open-telemetry:mainfrom
shashank-reddy-nr:feature/kafka-cluster-id

shashank-reddy-nr commented Jun 22, 2026 •

edited

Loading

Uh oh!

linux-foundation-easycla Bot commented Jun 22, 2026 •

edited

Loading

Uh oh!

shashank-reddy-nr commented Jul 15, 2026

Uh oh!

emdneto left a comment

Uh oh!

emdneto Aug 3, 2026

Uh oh!

emdneto Aug 3, 2026

Uh oh!

emdneto Aug 4, 2026

Uh oh!

emdneto Aug 4, 2026

Uh oh!

shashank-reddy-nr Aug 4, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Uh oh!

Conversation

shashank-reddy-nr commented Jun 22, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Type of change

How Has This Been Tested?

Does This PR Require a Core Repo Change?

Checklist:

Uh oh!

linux-foundation-easycla Bot commented Jun 22, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

shashank-reddy-nr commented Jul 15, 2026

Uh oh!

emdneto left a comment

Choose a reason for hiding this comment

Uh oh!

emdneto Aug 3, 2026

Choose a reason for hiding this comment

Uh oh!

emdneto Aug 3, 2026

Choose a reason for hiding this comment

Uh oh!

emdneto Aug 4, 2026

Choose a reason for hiding this comment

Uh oh!

emdneto Aug 4, 2026

Choose a reason for hiding this comment

Uh oh!

shashank-reddy-nr Aug 4, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

shashank-reddy-nr commented Jun 22, 2026 •

edited

Loading

linux-foundation-easycla Bot commented Jun 22, 2026 •

edited

Loading