Fix multicast context propagation race in reactor/reactive-streams

[amarziali]

What Does This Do

This change fixes a race in Reactor context propagation when multiple threads concurrently subscribe to the same shared publisher, such as a multicast sink returned by connection.asFlux().

Previously, the Reactor subscribe-time handoff used a shared (Publisher, Context) slot. Since the publisher instance can be shared across concurrent subscribers, one subscribing thread could accidentally consume a context deposited by another thread. When that happened, the affected subscriber stored the wrong parent context, and later onNext restored that incorrect context, causing spans to be attached to the wrong context.

The fix changes the Reactor bridge handoff to use a HandoffContext that is stamped with the producing thread. Consumers now only adopt the publisher handoff when it was produced on the same thread. If the handoff came from a different thread, it is treated as foreign and ignored, and the subscriber falls back to its currently active context.

Ignoring a foreign-thread handoff is safe for this Reactor bridge because the handoff is only intended to bridge context within a single synchronous subscribe flow. In that flow, the producer and the legitimate consumer run on the same thread.

Note: this confinement is intentionally applied only to the Reactor bridge. Other producers, such as resilience4j, Spring WebFlux, or Spring Messaging, may create context during assembly and subscribe later on another thread. For those integrations, cross-thread handoff can be legitimate, so they continue to use anyThread.

Motivation

Additional Notes

A good way to choose between anyThread and threadConfined is to ask: Can this same publisher be subscribed to by multiple threads at the same time?

If yes, use threadConfined.

That means the publisher is shared. Since several subscribers may use the same publisher slot concurrently, we must make sure a subscriber only uses context created by its own thread. This prevents one subscriber from accidentally picking up another subscriber’s context.

If no, use anyThread.

That means the publisher is specific to one operation. There is no competing subscriber that could steal the context. However, the subscription may still happen later on a different thread, so the context must remain usable across threads.

Contributor Checklist

• Format the title according to the contribution guidelines
• Assign the type: and (comp: or inst:) labels in addition to any other useful labels
• Avoid using 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, migration, or deletion
• Update public documentation with any new configuration flags or behaviors
• Add your completed PR to the merge queue by commenting /merge. You can also:
  • Customize the commit message associated with the merge with /merge --commit-message "..."
  • Remove your PR from the merge queue with /merge -c
  • Skip all merge queue checks with /merge -f --reason "reason"; please use this judiciously, as some checks do not run at the PR-level (note: the PR still needs to be mergeable, this will only skip the pre-merge build)
  • Get more information in this doc

Jira ticket: [PROJ-IDENT]

[amarziali]

@codex review

[chatgpt-codex-connector]

Codex Review: Didn't find any major issues. Hooray!

Reviewed commit: 08cc9ac832

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".

[datadog-prod-us1-6]

🎯 Code Coverage (details)
• Patch Coverage: 0.00%
• Overall Coverage: 54.42% (-0.60%)

_{This comment will be updated automatically if new data arrives.
🔗 Commit SHA: 08cc9ac | Docs | Datadog PR Page | Give us feedback!}

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 08cc9ac832

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".