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]>

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]>

… TrackKafkaProduceOffsetWithCluster params

FetchClusterIDAsync now accepts a cancellable context so Close() can
signal the fetch to stop early rather than blocking unconditionally.
Also reorders TrackKafkaProduceOffsetWithCluster params to match the
Kafka hierarchy: cluster, topic, partition, offset.

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

…formatting

Move cluster param to first position to match hierarchy (cluster, group,
topic, partition, offset). Fix struct field alignment in Tracer.

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

Cluster ID should always be available when running against a real Kafka
broker. Make assertions strict instead of conditionally skipping.

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

Deduplicate the admin client cluster ID fetch logic shared between
fetchClusterIDFromConsumer and fetchClusterIDFromProducer.

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

…tting

Replace separate CancelClusterIDFetch + WaitForClusterID calls in Close
with a single StopClusterIDFetch method. Fix gofmt struct alignment.

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

Remove the FetchClusterIDAsync/StopClusterIDFetch/WaitForClusterID
abstraction on the Tracer. Instead, use a standalone admin client
(kafka.NewAdminClient) so the goroutine has no shared state with the
consumer/producer and needs no coordination on Close(). Also remove
the cluster ID cache (clusterIDFromConfigOrFetch, cluster_id_cache.go).

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

…cluster ID

NewAdminClient does not strip go.* prefixed keys (unlike NewConsumer/
NewProducer), causing errors like "No such configuration property:
go.delivery.reports". Filter them out before creating the admin client,
preserving all auth/TLS/SASL config.

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

#4531)

## Summary

- Replace standalone goroutine fields
(`clusterIDCancel`/`clusterIDDone`) with a `closeAsync []func()` slice
on both `Consumer` and `Producer`
- Extract `startClusterIDFetch` helper that launches the cluster ID
fetch goroutine and returns a cancel+wait closure — context cancellation
unblocks the `ClusterID` call immediately on `Close()`
- Switch from standalone `kafka.NewAdminClient` to derived
`kafka.NewAdminClientFromConsumer`/`NewAdminClientFromProducer` —
eliminates `go.*` config key filtering
- Apply identical changes to both `kafka/` (v1) and `kafka.v2/` (v2)
- Port missing test coverage to v1: add
`TestConsumerFunctionalWithClusterID` and assert
`ext.MessagingKafkaClusterID` span tag in `TestConsumerFunctional`

To merge in #4470 to fix CI leak

The consumer's WrapConsumer was unconditionally spawning a goroutine to
fetch the Kafka cluster ID, even when data streams monitoring is
disabled. The producer's WrapProducer had the fetch outside the
DSMEnabled() block in the v2 variant. Wrap both consumer and producer
cluster ID fetches in DSMEnabled() guards across both kafka v1 and v2
packages.

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