refactor(dsm): remove WithClusterID option from kafka-go

Remove the public WithClusterID option since cluster ID is now
auto-detected. Add Close() to KafkaWriter to wait for async fetch.
Add WaitForClusterID calls in test helper for deterministic behavior.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: robcarlan-datadog

contrib/segmentio/kafka-go/internal/tracing/tracer.go

@@ -96,14 +96,6 @@ func WithDataStreams() Option {
 	})
 }
 
-// WithClusterID sets the Kafka cluster ID for Data Streams monitoring and span tagging.
-// This overrides the automatically detected cluster ID.
-func WithClusterID(clusterID string) Option {
-	return OptionFn(func(tr *Tracer) {
-		tr.kafkaCfg.ClusterID = clusterID
-	})
-}
-
 func (tr *Tracer) ClusterID() string {
 	tr.clusterIDMu.RLock()
 	defer tr.clusterIDMu.RUnlock()

contrib/segmentio/kafka-go/kafka.go

@@ -178,6 +178,12 @@ func parseAddrs(addr net.Addr) []string {
 	return strings.Split(addr.String(), ",")
 }
 
+// Close calls the underlying Writer.Close.
+func (w *KafkaWriter) Close() error {
+	w.tracer.WaitForClusterID()
+	return w.Writer.Close()
+}
+
 // WriteMessages calls kafka-go.Writer.WriteMessages and traces the requests.
 func (w *KafkaWriter) WriteMessages(ctx context.Context, msgs ...kafka.Message) error {
 	// although there's only one call made to the SyncProducer, the messages are

contrib/segmentio/kafka-go/kafka_test.go

@@ -162,11 +162,13 @@ func genIntegrationTestSpans(t *testing.T, mt mocktracer.Tracer, writerOp func(t
 		writtenMessages = append(writtenMessages, messages...)
 	}
 	w := WrapWriter(kw, writerOpts...)
+	w.tracer.WaitForClusterID()
 	writerOp(t, w)
 	err := w.Close()
 	require.NoError(t, err)
 
 	r := WrapReader(testReader(), readerOpts...)
+	r.tracer.WaitForClusterID()
 	readerOp(t, r)
 	err = r.Close()
 	require.NoError(t, err)

contrib/segmentio/kafka-go/option.go

@@ -34,8 +34,3 @@ func WithDataStreams() Option {
 	return tracing.WithDataStreams()
 }
 
-// WithClusterID sets the Kafka cluster ID for Data Streams monitoring and span tagging.
-// This overrides the automatically detected cluster ID.
-func WithClusterID(clusterID string) Option {
-	return tracing.WithClusterID(clusterID)
-}