fix(v3): prevent SERVER span context leak in Kafka consumer; add SpanLinks

The Kafka consumer batch handler was inheriting the active Context.current()
when building its CONSUMER span. On Vert.x's event loop, this meant the last
HTTP SERVER span's context leaked into consumer spans, making them appear as
children of unrelated HTTP traces.

Fixes:
- Use Context.root() as explicit parent in tracedBatchHandler so consumer
  spans are always root spans, independent of any leaked HTTP context
- Extract traceparent from each Kafka record's headers and add as SpanLink,
  connecting the consumer span to its producer span(s) per OTel messaging
  semconv (async decoupled relationship → SpanLink, not parent/child)

Tests added:
- consumerSpanIsRootSpanEvenWhenServerSpanIsActive: verifies Context.root()
  prevents SERVER span context leak
- consumerSpanLinksToProducerViaTraceparentHeader: verifies traceparent header
  extraction and SpanLink creation

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: prathamesh-sonpatki

vertx3-rxjava2-otel-autoconfigure/src/main/java/io/last9/tracing/otel/v3/KafkaTracing.java

@@ -4,22 +4,30 @@
 import io.opentelemetry.api.OpenTelemetry;
 import io.opentelemetry.api.common.Attributes;
 import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.SpanContext;
 import io.opentelemetry.api.trace.SpanKind;
 import io.opentelemetry.api.trace.StatusCode;
 import io.opentelemetry.api.trace.Tracer;
 import io.opentelemetry.context.Context;
 import io.opentelemetry.context.Scope;
+import io.opentelemetry.context.propagation.TextMapGetter;
 import io.opentelemetry.context.propagation.TextMapSetter;
 import io.opentelemetry.semconv.ExceptionAttributes;
 import io.opentelemetry.semconv.SemanticAttributes;
 import io.reactivex.Single;
 import io.vertx.core.Handler;
+import io.vertx.kafka.client.consumer.KafkaConsumerRecord;
 import io.vertx.kafka.client.consumer.KafkaConsumerRecords;
+import io.vertx.kafka.client.producer.KafkaHeader;
 import io.vertx.kafka.client.producer.RecordMetadata;
 import io.vertx.reactivex.kafka.client.consumer.KafkaConsumer;
 import io.vertx.reactivex.kafka.client.producer.KafkaProducer;
 import io.vertx.reactivex.kafka.client.producer.KafkaProducerRecord;
 
+import java.nio.charset.StandardCharsets;
+import java.util.Iterator;
+import java.util.List;
+
 /**
  * Utility for adding OpenTelemetry tracing to Vert.x 3 Kafka producers and consumers.
  *
@@ -63,6 +71,35 @@ public final class KafkaTracing {
     private static final TextMapSetter<KafkaProducerRecord> RECORD_SETTER =
             (record, key, value) -> record.addHeader(key, value);
 
+    /**
+     * {@link TextMapGetter} that extracts trace context from Kafka consumer record headers.
+     * Used to extract the producer's {@code traceparent} and add it as a
+     * {@link io.opentelemetry.api.trace.SpanContext} link on the CONSUMER span.
+     */
+    @SuppressWarnings({"rawtypes", "unchecked"})
+    private static final TextMapGetter<KafkaConsumerRecord> RECORD_GETTER =
+            new TextMapGetter<KafkaConsumerRecord>() {
+                @Override
+                public Iterable<String> keys(KafkaConsumerRecord record) {
+                    List<String> keys = new java.util.ArrayList<>();
+                    for (KafkaHeader h : (List<KafkaHeader>) record.headers()) {
+                        keys.add(h.key());
+                    }
+                    return keys;
+                }
+
+                @Override
+                public String get(KafkaConsumerRecord record, String key) {
+                    if (record == null) return null;
+                    for (KafkaHeader h : (List<KafkaHeader>) record.headers()) {
+                        if (key.equals(h.key())) {
+                            return new String(h.value().getBytes(), StandardCharsets.UTF_8);
+                        }
+                    }
+                    return null;
+                }
+            };
+
     private KafkaTracing() {
         // Utility class
     }
@@ -129,14 +166,20 @@ public static <K, V> Handler<KafkaConsumerRecords<K, V>> tracedBatchHandler(
      * @param openTelemetry  the OpenTelemetry instance to use
      * @return a wrapped handler that creates and ends a CONSUMER span around each batch
      */
+    @SuppressWarnings({"unchecked", "rawtypes"})
     public static <K, V> Handler<KafkaConsumerRecords<K, V>> tracedBatchHandler(
             String topic,
             String consumerGroup,
             Handler<KafkaConsumerRecords<K, V>> delegate,
             OpenTelemetry openTelemetry) {
         Tracer tracer = openTelemetry.getTracer(TRACER_NAME);
         return records -> {
+            // Use Context.root() as parent to prevent leaked HTTP SERVER span context from
+            // polluting Kafka consumer spans (same fix as TracedRouter for HTTP handlers).
+            // Consumer spans are root spans per OTel messaging semconv — they link to producers
+            // via SpanLink rather than parent/child to model the async relationship accurately.
             var spanBuilder = tracer.spanBuilder(topic + " process")
+                    .setParent(Context.root())
                     .setSpanKind(SpanKind.CONSUMER)
                     .setAttribute(SemanticAttributes.MESSAGING_SYSTEM, "kafka")
                     .setAttribute(SemanticAttributes.MESSAGING_DESTINATION_NAME, topic)
@@ -150,6 +193,19 @@ public static <K, V> Handler<KafkaConsumerRecords<K, V>> tracedBatchHandler(
                         consumerGroup);
             }
 
+            // Add a SpanLink for each record that carries a traceparent header.
+            // This connects the consumer span to the producer span(s) without making it
+            // a child — correct for async messaging where producer and consumer are decoupled.
+            for (int i = 0; i < records.size(); i++) {
+                KafkaConsumerRecord record = records.recordAt(i);
+                Context producerCtx = openTelemetry.getPropagators().getTextMapPropagator()
+                        .extract(Context.root(), record, RECORD_GETTER);
+                SpanContext producerSpanCtx = Span.fromContext(producerCtx).getSpanContext();
+                if (producerSpanCtx.isValid()) {
+                    spanBuilder.addLink(producerSpanCtx);
+                }
+            }
+
             Span span = spanBuilder.startSpan();
             try (Scope ignored = span.makeCurrent()) {
                 delegate.handle(records);

vertx3-rxjava2-otel-autoconfigure/src/test/java/io/last9/tracing/otel/v3/KafkaTracingTest.java

@@ -4,6 +4,7 @@
 import io.opentelemetry.api.trace.Span;
 import io.opentelemetry.api.trace.SpanKind;
 import io.opentelemetry.api.trace.StatusCode;
+import io.opentelemetry.context.Scope;
 import io.opentelemetry.sdk.testing.exporter.InMemorySpanExporter;
 import io.opentelemetry.sdk.trace.data.SpanData;
 import io.reactivex.Single;
@@ -19,6 +20,7 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
+import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -294,6 +296,85 @@ void producerSendTombstoneSetsMsgTombstoneAttribute() {
                 .isTrue();
     }
 
+    // ---- Context isolation tests ----
+
+    @Test
+    void consumerSpanIsRootSpanEvenWhenServerSpanIsActive() {
+        // Reproduce the SERVER span context leak: a Vert.x HTTP SERVER span is active
+        // on the event loop thread. Without Context.root() as parent, the consumer span
+        // would become a child of that SERVER span — producing a wrong trace.
+        Span serverSpan = otel.getOpenTelemetry().getTracer("test")
+                .spanBuilder("GET /api/data")
+                .setSpanKind(SpanKind.SERVER)
+                .startSpan();
+
+        try (Scope ignored = serverSpan.makeCurrent()) {
+            Handler<KafkaConsumerRecords<String, String>> traced = KafkaTracing.tracedBatchHandler(
+                    "orders",
+                    records -> {},
+                    otel.getOpenTelemetry()
+            );
+            traced.handle(emptyBatch());
+        } finally {
+            serverSpan.end();
+        }
+
+        List<SpanData> spans = spanExporter.getFinishedSpanItems();
+        SpanData consumerSpan = spans.stream()
+                .filter(s -> s.getKind() == SpanKind.CONSUMER)
+                .findFirst().orElseThrow();
+
+        // Consumer span must be a root span — not a child of the HTTP SERVER span
+        assertThat(consumerSpan.getParentSpanContext().isValid()).isFalse();
+    }
+
+    @Test
+    void consumerSpanLinksToProducerViaTraceparentHeader() {
+        // Simulate a producer span with a known trace/span ID
+        Span producerSpan = otel.getOpenTelemetry().getTracer("test")
+                .spanBuilder("orders publish")
+                .setSpanKind(SpanKind.PRODUCER)
+                .startSpan();
+        String producerTraceId = producerSpan.getSpanContext().getTraceId();
+        String producerSpanId = producerSpan.getSpanContext().getSpanId();
+        producerSpan.end();
+
+        // Build a W3C traceparent header value (same format TracedKafkaProducer injects)
+        String traceparent = "00-" + producerTraceId + "-" + producerSpanId + "-01";
+
+        // Create a Kafka ConsumerRecord carrying the traceparent header
+        ConsumerRecord<String, String> rawRecord =
+                new ConsumerRecord<>("orders", 0, 0L, "key-1", "value-1");
+        rawRecord.headers().add("traceparent", traceparent.getBytes(StandardCharsets.UTF_8));
+
+        ConsumerRecords<String, String> raw = new ConsumerRecords<>(
+                Collections.singletonMap(new TopicPartition("orders", 0),
+                        Collections.singletonList(rawRecord)));
+
+        Handler<KafkaConsumerRecords<String, String>> traced = KafkaTracing.tracedBatchHandler(
+                "orders",
+                records -> {},
+                otel.getOpenTelemetry()
+        );
+        traced.handle(new KafkaConsumerRecordsImpl<>(raw));
+
+        List<SpanData> spans = spanExporter.getFinishedSpanItems();
+        // Only the consumer span — producer span ended before the handler ran
+        SpanData consumerSpan = spans.stream()
+                .filter(s -> s.getKind() == SpanKind.CONSUMER)
+                .findFirst().orElseThrow();
+
+        // Consumer span must be a root span (no parent)
+        assertThat(consumerSpan.getParentSpanContext().isValid()).isFalse();
+
+        // Consumer span must carry a SpanLink pointing to the producer span
+        assertThat(consumerSpan.getLinks()).hasSize(1);
+        assertThat(consumerSpan.getLinks().get(0).getSpanContext().getTraceId())
+                .isEqualTo(producerTraceId);
+        assertThat(consumerSpan.getLinks().get(0).getSpanContext().getSpanId())
+                .isEqualTo(producerSpanId);
+    }
+
     // ---- Helpers ----
 
     private KafkaConsumerRecords<String, String> emptyBatch() {