Skip to content

Add RxJava 3 instrumentation#11506

Draft
amarziali wants to merge 10 commits into
masterfrom
andrea.marziali/rxjava3
Draft

Add RxJava 3 instrumentation#11506
amarziali wants to merge 10 commits into
masterfrom
andrea.marziali/rxjava3

Conversation

@amarziali
Copy link
Copy Markdown
Contributor

What Does This Do

We already trace RxJava 1 and 2 but not 3. RxJava 3 moved its types to the io.reactivex.rxjava3.core.* namespace, so the existing rxjava-2.0 instrumentation never matches it. This adds a new rxjava-3.0 module that brings RxJava 3 to parity.

What's in it:

  • New dd-java-agent/instrumentation/rxjava/rxjava-3.0 module. It ports the rxjava-2.0 logic to the RxJava 3 namespace: context capture on the five reactive types (Flowable, Observable, Single, Maybe, Completable) and re-attachment around subscriber callbacks, plus the async result extension that finishes @WithSpan / @Trace spans when the stream completes, errors, or is cancelled.
  • Muzzle has a pass for rxjava3 [3.0.0,) and a fail block that asserts the advice can never match the rxjava2 artifact, so the two versions stay isolated. The module also pulls in rxjava-2.0 as a test runtime dependency to confirm both instrumenters coexist.
  • Registered in the GraalVM native image build time list next to the rxjava2 entry.
  • Tests are JUnit 5 (44 cases): subscription propagation, the core propagation suite, the @WithSpan result extension across all five types, and a Java 8 interop check (fromCompletionStage, fromStream, fromOptional). latestDepTest runs the same suite against the newest published rxjava3.

One thing reviewers should look at closely: this also touches the shared java-concurrent module. While testing delayed chains I found that RxJava 3 added io.reactivex.rxjava3.internal.schedulers.AbstractDirectTask, whose static initializer builds two sentinel FutureTask instances. When that initializer first runs under an active trace (the first delay or timeout hop inside a span), the executor instrumentation captures a continuation on those static singletons that never gets cancelled, so the trace stays pending and is never reported. RxJava 2 has no equivalent class, which is why the byte for byte equivalent rxjava-2.0 code never hit this. The fix disables async propagation while that type initializer runs, following the same pattern already used for Reactor's SchedulerTask and WorkerTask. The matcher is an exact class name, so it stays inert unless RxJava 3 is on the classpath.

The Java 8 interop check found no propagation gaps once that fix was in place.

Motivation

Additional Notes

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
    • Get more information in this doc

Jira ticket: [PROJ-IDENT]

amarziali and others added 10 commits May 29, 2026 17:42
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…izer

RxJava 3 introduced io.reactivex.rxjava3.internal.schedulers.AbstractDirectTask,
whose static initializer constructs FINISHED/DISPOSED sentinel FutureTask
instances. When that initializer first runs under an active trace (e.g. the
first delay/timeout scheduler hop inside a span), the executor instrumentation
captures a ScopeContinuation on those static singletons that is never cancelled,
leaking the pending trace so it is never reported.

Disable async propagation while AbstractDirectTask's type initializer runs,
mirroring the existing reactor.core.scheduler.SchedulerTask/WorkerTask handling.
The matcher is inert unless RxJava 3 is on the classpath. RxJava 2 has no
equivalent class. Restores the delayed-Maybe coverage in RxJava3Test, which
fails without this fix and passes with it.

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

Two tests were under datadog.trace.instrumentation.rxjava3 while the @Trace-using
tests must live under testdog.* (the agent ignores datadog.* for instrumentation).
Move all four tests to testdog.* for consistency.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
@amarziali amarziali added type: enhancement Enhancements and improvements inst: others All other instrumentations ai-generated AI generated code labels May 29, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

ai-generated AI generated code inst: others All other instrumentations type: enhancement Enhancements and improvements

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant