Extract trace context from Kafka producer record headers

[PlugaruT]

What Does This Do

Enables Kafka producers to extract and continue existing traces from record headers.

Motivation

Fix for #5092

Currently, when a Kafka producer sends a record that already has trace context in its headers (e.g., when forwarding messages between services), the tracer creates a new root span instead of continuing the existing trace. This breaks distributed tracing continuity

This change extracts the trace context from record headers using extractContextAndGetSpanContext() and uses it as the parent for the produce span, maintaining trace continuity across service boundaries. It's also making dd-trace behave the same way opentelemetry does, see here.

Additional Notes

Contributor Checklist

• Format the title according the contribution guidelines
• Assign the type: and (comp: or inst:) labels in addition to any useful labels
• Don't use close, fix or any linking keywords when referencing an issue.
  Use solves instead, and assign the PR milestone to the issue
• Update the CODEOWNERS file on source file addition, move, or deletion
• Update the public documentation in case of new configuration flag or behavior

Jira ticket: [PROJ-IDENT]

[PerfectSlayer]

Hi @PlugaruT 👋 Thanks for taking time to contribute.

Quick feedback about the change but I think this might break the context tracking later as KafkaProducerCallback won't use the extracted context.

What I would do is:

• Declare an AgentSpanContext
• Try to extract from record headers
• If null, set it with active span
• Reuse this span context for both startSpan and KafkaProducerCallback.

I will send the current PR into CI to get more feedback (see #10022) and ping @DataDog/apm-idm-java to get a proper review

[PlugaruT]

Hi @PlugaruT 👋 Thanks for taking time to contribute.

Quick feedback about the change but I think this might break the context tracking later as KafkaProducerCallback won't use the extracted context.

What I would do is:

* Declare an AgentSpanContext

* Try to extract from record headers

* If null, set it with active span

* Reuse this span context for both `startSpan` and `KafkaProducerCallback`.

I will send the current PR into CI to get more feedback (see #10022) and ping @DataDog/apm-idm-java to get a proper review

Hi @PerfectSlayer, are you referring to how parent is passed down to the KafkaProducerCallback? Seems that indeed it's a flaw in my logic, for some reason I was under the impression that it's automatically passed with the span. My miss. Will get it fixed.

[PerfectSlayer]

are you referring to how parent is passed down to the KafkaProducerCallback?

Yes

Seems that indeed it's a flaw in my logic, for some reason I was under the impression that it's automatically passed with the span. My miss. Will get it fixed.

The active span content is implicitly used for startSpan() if you don’t set one, but the KakfaProducerCallback need an explicit one. So better declare one explicit parent context and use it for both.

[PlugaruT]

Hey @DataDog/apm-idm-java, can I get a review on this one 🙏

[PerfectSlayer]

I pinged the team in their internal Slack channel

[ygree]

This could be a breaking change if users expect activeSpan to take precedence. Adding a configuration option, would allow users to configure this behavior.

[PlugaruT]

I actually was expecting the other way around, especially coming from opentelemetry, I expected to have the context reuse the one from headers and attach new spans to the same trace and not break it.
I can do it, no problem there, but I'd say this is a bug fix rather than a feature.

[PlugaruT]

It's done. Let me know if the name works

[ygree]

Makes sense. I'll exclude the commit that adds the setting, then. Let's see if all the CI checks pass in the mirroring PR: #10022

[PlugaruT]

I'm looking into the failing test. Just found how to run the system tests locally.

[PlugaruT]

@ygree I force pushed without the commit that added the config and also fixed the test, at least it passes on my computer 😅

[ygree]

Thanks! I'll pull your changes to the mirroring PR and run the CI checks.

[ygree]

The same change should be made in datadog.trace.instrumentation.kafka_clients38.ProducerAdvice to cover Kafka 3.8+ clients.

[ygree]

@PlugaruT Thank you for your contribution! Mirror PR #10022 has been merged.