[FEATURE]: Activate each trace from a Kafka batch

[PlugaruT]

Package Name

kafka

Package Version(s)

No response

Describe the goal of the feature

The current implementation is activating the first trace from the first message in a given batch, see

dd-trace-py/ddtrace/contrib/internal/kafka/patch.py

Line 247 in 73764dc

first_message = messages[0] if len(messages) else None

Is it possible to change this behaviour in order to activate each trace individually from a given batch? Currently we have a lot of spans attached to a trace that does not correspond to the correct source Kafka record, hence causing confusion. In Java SDK, the trace is activated for each Kafka record.

Or at least, what's the recommendation in order to achieve this behaviour?

Is your feature request related to a problem?

No response

Describe alternatives you've considered

No response

Additional context

No response

[PlugaruT]

OTel seems to at least create a link between them
https://github.com/open-telemetry/opentelemetry-python-contrib/blob/main/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py#L104 so maybe consider that?

Open to contribute if we agree on the implementation

[PlugaruT]

I think a patch would look like this:

From 0000000000000000000000000000000000000000 Mon Sep 17 00:00:00 2001
From: Your Name <you@example.com>
Date: Fri, 8 May 2026 12:00:00 +0000
Subject: [PATCH] kafka: link all producer traces in batched consume

---
 ddtrace/contrib/internal/kafka/patch.py | 23 +++++++++++++++++------
 1 file changed, 17 insertions(+), 6 deletions(-)

diff --git a/ddtrace/contrib/internal/kafka/patch.py b/ddtrace/contrib/internal/kafka/patch.py
--- a/ddtrace/contrib/internal/kafka/patch.py
+++ b/ddtrace/contrib/internal/kafka/patch.py
@@ -244,20 +244,31 @@ def traced_poll_or_consume(func, instance, args, kwargs):
 
 
 def _instrument_message(messages, pin, start_ns, instance, err):
-    ctx = None
-    # First message is used to extract context and enrich datadog spans
-    # This approach aligns with the opentelemetry confluent kafka semantics
-    first_message = messages[0] if len(messages) else None
-    if first_message is not None and config.kafka.distributed_tracing_enabled and first_message.headers():
-        ctx = Propagator.extract(dict(first_message.headers()))
+    # Extract distributed tracing context from every message in the batch.
+    # The first message's context is used to parent the consume span
+    # (preserving prior behavior for single-message poll() traces). Any
+    # additional contexts are attached as span links so that batched
+    # consumes do not orphan the producer traces of messages 2..N.
+    contexts = []
+    if config.kafka.distributed_tracing_enabled:
+        for message in messages:
+            if message is None or not message.headers():
+                continue
+            ctx = Propagator.extract(dict(message.headers()))
+            if ctx is not None and ctx.trace_id is not None:
+                contexts.append(ctx)
+
+    first_message = messages[0] if len(messages) else None
 
     with pin.tracer.start_span(
         name=schematize_messaging_operation(kafkax.CONSUME, provider="kafka", direction=SpanDirection.PROCESSING),
         service=trace_utils.ext_service(pin, config.kafka),
         span_type=SpanTypes.WORKER,
-        child_of=ctx if ctx is not None and ctx.trace_id is not None else pin.tracer.context_provider.active(),
+        child_of=contexts[0] if contexts else pin.tracer.context_provider.active(),
         activate=True,
     ) as span:
+        # Link the remaining producer traces in the batch so they are not
+        # orphaned by the single-parent consume span semantics above.
+        for link_ctx in contexts[1:]:
+            span.link_span(link_ctx)
         # reset span start time to before function call
         span.start_ns = start_ns
 
--