feat(dsm): add kafka_cluster_id to confluent-kafka-go by robcarlan-datadog · Pull Request #4470 · DataDog/dd-trace-go

robcarlan-datadog · 2026-02-23T21:04:52Z

Summary

Most DSM tracer instrumentations for Kafka enrich a kafka_cluster_id. This is useful because numerous generated metrics such as data_streams.latency and data_streams.kafka.lag_messages will automatically add this metric if present.
In situations where people have the same topic names across multiple kafka clusters, the above metric will become incorrect without the cluster id tag. This PR adds feature parity for Go and collects this metric tag when DSM is enabled.

This is implemented by:

Launching a goroutine on consumer/producer creation to initiate the admin API and query for cluster id
Enriching metrics with this cluster id after we obtain it
Signalling a stop for an inflight cluster id request if the consumer/producer closes early
Enabling this only for DSM

For context, the other tracers implement this as follows:

Java (doesn't block, intercepts the metadata response to enrich cluster id going forwards, see here and here)
Node (blocks, see here and here)
Python (blocks with a 1 second timeout, see here and here)

Test plan

Add to existing tests
Test manually
Screenshots:

Spans have kafka_cluster_id:

Metric now adds kafka_cluster_id:

🤖 Generated with Claude Code

datadog-datadog-prod-us1 · 2026-02-25T18:53:55Z

✅ Tests

🎉 All green!

❄️ No new flaky tests detected
🧪 All tests passed

🎯 Code Coverage (details)
• Patch Coverage: 60.00%
• Overall Coverage: 59.38% (+3.53%)

_{This comment will be updated automatically if new data arrives.

🔗 Commit SHA: b893370 | Docs | Datadog PR Page | Was this helpful? React with 👍/👎 or give us feedback!}

pr-commenter · 2026-02-25T19:08:11Z

Benchmarks

Benchmark execution time: 2026-03-19 19:23:04

Comparing candidate commit b893370 in PR branch rob.carlan/dsm-kafka-cluster-id with baseline commit 0b8e268 in branch main.

Found 0 performance improvements and 0 performance regressions! Performance is the same for 157 metrics, 7 unstable metrics.

Explanation

This is an A/B test comparing a candidate commit's performance against that of a baseline commit. Performance changes are noted in the tables below as:

🟩 = significantly better candidate vs. baseline
🟥 = significantly worse candidate vs. baseline

We compute a confidence interval (CI) over the relative difference of means between metrics from the candidate and baseline commits, considering the baseline as the reference.

If the CI is entirely outside the configured SIGNIFICANT_IMPACT_THRESHOLD (or the deprecated UNCONFIDENCE_THRESHOLD), the change is considered significant.

Feel free to reach out to #apm-benchmarking-platform on Slack if you have any questions.

More details about the CI and significant changes

You can imagine this CI as a range of values that is likely to contain the true difference of means between the candidate and baseline commits.

CIs of the difference of means are often centered around 0%, because often changes are not that big:

---------------------------------(------|---^--------)-------------------------------->
                              -0.6%    0%  0.3%     +1.2%
                                 |          |        |
         lower bound of the CI --'          |        |
sample mean (center of the CI) -------------'        |
         upper bound of the CI ----------------------'

As described above, a change is considered significant if the CI is entirely outside the configured SIGNIFICANT_IMPACT_THRESHOLD (or the deprecated UNCONFIDENCE_THRESHOLD).

For instance, for an execution time metric, this confidence interval indicates a significantly worse performance:

----------------------------------------|---------|---(---------^---------)---------->
                                       0%        1%  1.3%      2.2%      3.1%
                                                  |   |         |         |
       significant impact threshold --------------'   |         |         |
                      lower bound of CI --------------'         |         |
       sample mean (center of the CI) --------------------------'         |
                      upper bound of CI ----------------------------------'

codecov · 2026-02-25T20:42:44Z

Codecov Report

❌ Patch coverage is 50.00000% with 15 lines in your changes missing coverage. Please review.
✅ Project coverage is 60.05%. Comparing base (0b8e268) to head (b893370).

Files with missing lines	Patch %	Lines
internal/datastreams/processor.go	62.50%	9 Missing ⚠️
ddtrace/tracer/data_streams.go	0.00%	6 Missing ⚠️

Additional details and impacted files

Files with missing lines	Coverage Δ
internal/datastreams/pathway.go	`86.11% <ø> (ø)`
ddtrace/tracer/data_streams.go	`33.33% <0.00%> (ø)`
internal/datastreams/processor.go	`62.56% <62.50%> (ø)`

... and 431 files with indirect coverage changes

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

Plumb Kafka cluster ID through the DSM pipeline so that offsets and checkpoints from different clusters are distinguishable. For confluent-kafka-go, the cluster ID is fetched automatically via the AdminClient API; other Kafka libraries pass empty string for now and will be wired in follow-up PRs. Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>

Add cluster ID as a span tag on produce and consume spans, add ext constant MessagingKafkaClusterID, and add integration test verifying cluster ID flows into spans, DSM pathway hashes, and backlog tags. Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>

Auto-fetched cluster ID from the Kafka admin API is now prepended to options so that user-provided WithClusterID takes precedence. Also fix TestConsumerFunctional pathway hash to account for auto-fetched cluster ID in integration test environments. Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>

Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>

The TestConsumerFunctional test was computing expected pathway hashes without the kafka_cluster_id tag, but the production code now includes it after auto-fetching from the broker. Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>

Avoid repeated admin API calls to fetch cluster ID when creating multiple consumers/producers to the same Kafka cluster. The cache is shared between v1 and v2 packages via kafkatrace. Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>

Remove the public WithClusterID option function which was only used internally and for testing. Set cluster ID directly on the tracer after wrapping instead. Also reduce cluster ID fetch timeout from 10s to 2s. Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>

mtoffl01

Other changes look fine, but my real concern is the blocking network call. I think we should prefer to leave cluster ID out until we have a response from the network. This can be achieved with goroutine/channel to allow async behavior > synchronous

Move cluster ID fetching to a background goroutine so it doesn't block Consumer/Producer creation. Protect the clusterID field with a RWMutex for safe concurrent access. Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>

Move cluster ID goroutine lifecycle into the Tracer via FetchClusterIDAsync/WaitForClusterID. Close() waits for the goroutine to prevent use-after-free of the librdkafka handle. Tests call WaitForClusterID before producing/consuming to ensure deterministic pathway hashes. Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>

darccio · 2026-03-18T10:33:57Z

@codex review

chatgpt-codex-connector

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: f5b176c985

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

Open a pull request for review
Mark a draft as ready
Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".

darccio

LGTM. Please check Codex's comment.

robcarlan-datadog · 2026-03-18T18:01:39Z

The failures are all TestTraceProtocol, not related to this PR

johannbotha

LGTM

Add kafka_cluster_id support to both IBM/sarama and Shopify/sarama, matching the pattern from the confluent-kafka-go integration (PR #4470): - cfg.ClusterID() / cfg.SetClusterID() with sync.RWMutex for thread-safe access from the message path and the background fetch goroutine - closeAsync []func() on wrapped producers for cancelling in-flight fetches on Close() - startClusterIDFetch launches a background goroutine that connects to a broker, sends a MetadataRequest v4, and calls cfg.SetClusterID() - WithBrokers(addrs) option provides broker addresses (required because sarama's producer/consumer interfaces don't expose broker info) Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>

) ## Summary - Adds `kafka_cluster_id` support to the `segmentio/kafka-go` DSM integration, building on #4470 - Auto-fetches the Kafka cluster ID via async metadata request in `WrapReader`/`WrapWriter` - Plumbs cluster ID through DSM edge tags, offset tracking, and span tags Tested this manually and via unit/integration tests. Co-authored-by: genesor <[email protected]> Co-authored-by: rob.carlan <[email protected]>

## Summary - This follows practically the same pattern as the [confluent-kafka-go](#4470) and [kafka-go](#4477) PRs. However, this integration requires the user to supply the bootstrap servers explicitly in order to obtain the cluster id. - Adds `WithBrokers` option to both IBM/sarama and Shopify/sarama integrations that fetches the Kafka cluster ID via a metadata request and includes it in DSM payloads (DSM only) - Tested manually and via unit / integration tests. Co-authored-by: genesor <[email protected]> Co-authored-by: rob.carlan <[email protected]>

robcarlan-datadog force-pushed the rob.carlan/dsm-kafka-cluster-id branch 2 times, most recently from 0c09dde to 7de417d Compare February 25, 2026 18:47

github-actions Bot added the apm:ecosystem contrib/* related feature requests or bugs label Feb 25, 2026

robcarlan-datadog and others added 6 commits February 26, 2026 12:11

chore: update api.txt for new DSM cluster ID functions

e0ad51d

Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>

fix tests

354ebc1

robcarlan-datadog force-pushed the rob.carlan/dsm-kafka-cluster-id branch from 702c277 to f2cb692 Compare February 26, 2026 17:12

robcarlan-datadog mentioned this pull request Feb 26, 2026

feat(dsm): add kafka_cluster_id to segmentio/kafka-go integration #4477

Merged

robcarlan-datadog marked this pull request as ready for review February 26, 2026 18:45

robcarlan-datadog requested review from a team as code owners February 26, 2026 18:45

robcarlan-datadog changed the title ~~feat(dsm): add kafka_cluster_id to Data Streams Monitoring~~ feat(dsm): add kafka_cluster_id to confluent-kafka-go Mar 2, 2026

robcarlan-datadog mentioned this pull request Mar 2, 2026

feat(dsm): add kafka_cluster_id to IBM/sarama integration #4486

Closed

3 tasks

remove comments

5514993

robcarlan-datadog added the AI Generated Largely based on code generated by an AI or LLM. This label is the same across all dd-trace-* repos label Mar 2, 2026

mtoffl01 reviewed Mar 2, 2026

View reviewed changes

Comment thread contrib/confluentinc/confluent-kafka-go/kafka.v2/kafka.go Outdated

Comment thread contrib/confluentinc/confluent-kafka-go/kafka.v2/kafka.go Outdated

robcarlan-datadog and others added 2 commits March 2, 2026 14:35

Merge branch 'main' into rob.carlan/dsm-kafka-cluster-id

388479b

robcarlan-datadog requested a review from mtoffl01 March 3, 2026 16:58

mtoffl01 reviewed Mar 3, 2026

View reviewed changes

Comment thread contrib/confluentinc/confluent-kafka-go/kafka.v2/kafka.go Outdated

Comment thread contrib/confluentinc/confluent-kafka-go/kafka.v2/kafka.go Outdated

Comment thread contrib/confluentinc/confluent-kafka-go/kafkatrace/cluster_id_cache.go Outdated

robcarlan-datadog and others added 2 commits March 4, 2026 10:32

chatgpt-codex-connector Bot reviewed Mar 18, 2026

View reviewed changes

Comment thread contrib/confluentinc/confluent-kafka-go/kafka/kafka.go

darccio approved these changes Mar 18, 2026

View reviewed changes

robcarlan-datadog added 2 commits March 18, 2026 12:49

dont log error on cancellation

b418700

Merge branch 'main' into rob.carlan/dsm-kafka-cluster-id

9568d67

johannbotha approved these changes Mar 18, 2026

View reviewed changes

Merge branch 'main' into rob.carlan/dsm-kafka-cluster-id

5ef1f76

gh-worker-dd-devflow-36fce6 Bot added the mergequeue-status: waiting label Mar 19, 2026

Merge branch 'main' into rob.carlan/dsm-kafka-cluster-id

c3831a5

gh-worker-dd-devflow-36fce6 Bot added mergequeue-status: removed mergequeue-status: waiting and removed mergequeue-status: waiting mergequeue-status: removed labels Mar 19, 2026

Merge branch 'main' into rob.carlan/dsm-kafka-cluster-id

b893370

gh-worker-dd-devflow-36fce6 Bot added mergequeue-status: removed mergequeue-status: queued mergequeue-status: in_progress and removed mergequeue-status: waiting mergequeue-status: removed mergequeue-status: queued labels Mar 19, 2026

gh-worker-dd-mergequeue-cf854d Bot merged commit 6ae9763 into main Mar 19, 2026
116 of 117 checks passed

gh-worker-dd-devflow-36fce6 Bot added mergequeue-status: done and removed mergequeue-status: in_progress labels Mar 19, 2026

gh-worker-dd-mergequeue-cf854d Bot deleted the rob.carlan/dsm-kafka-cluster-id branch March 19, 2026 23:38

robcarlan-datadog mentioned this pull request Mar 20, 2026

feat(dsm): add kafka_cluster_id to Sarama integrations #4560

Merged

Uh oh!

Conversation

robcarlan-datadog commented Feb 23, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Test plan

Uh oh!

datadog-datadog-prod-us1 Bot commented Feb 25, 2026 • edited by datadog-prod-us1-5 Bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

pr-commenter Bot commented Feb 25, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Benchmarks

Explanation

More details about the CI and significant changes

Uh oh!

codecov Bot commented Feb 25, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

Uh oh!

Uh oh!

mtoffl01 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

darccio commented Mar 18, 2026

Uh oh!

chatgpt-codex-connector Bot left a comment

Choose a reason for hiding this comment

💡 Codex Review

Uh oh!

Uh oh!

darccio left a comment

Choose a reason for hiding this comment

Uh oh!

robcarlan-datadog commented Mar 18, 2026

Uh oh!

johannbotha left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

6 participants

robcarlan-datadog commented Feb 23, 2026 •

edited

Loading

datadog-datadog-prod-us1 Bot commented Feb 25, 2026 •

edited by datadog-prod-us1-5 Bot

Loading

pr-commenter Bot commented Feb 25, 2026 •

edited

Loading

codecov Bot commented Feb 25, 2026 •

edited

Loading