DataDog · amarziali · May 29, 2026 · May 29, 2026 · May 29, 2026 · May 29, 2026
@@ -156,6 +156,7 @@ public static void onEnter(@Advice.Argument(value = 0, readOnly = false) String[
               + "datadog.trace.instrumentation.reactivestreams.ReactiveStreamsAsyncResultExtension:build_time,"
               + "datadog.trace.instrumentation.reactor.core.ReactorAsyncResultExtension:build_time,"
               + "datadog.trace.instrumentation.rxjava2.RxJavaAsyncResultExtension:build_time,"
+              + "datadog.trace.instrumentation.rxjava3.RxJavaAsyncResultExtension:build_time,"
               + "datadog.trace.logging.ddlogger.DDLogger:build_time,"
               + "datadog.trace.logging.ddlogger.DDLoggerFactory:build_time,"
               + "datadog.trace.logging.ddlogger.DDLoggerFactory$HelperWrapper:build_time,"

@@ -46,6 +46,16 @@ public AsyncPropagatingDisableInstrumentation() {
   private static final ElementMatcher<TypeDescription> REACTOR_DISABLED_TYPE_INITIALIZERS =
       namedOneOf("reactor.core.scheduler.SchedulerTask", "reactor.core.scheduler.WorkerTask");
 
+  /**
+   * RxJava 3's AbstractDirectTask creates FINISHED/DISPOSED sentinel FutureTask instances in its
+   * static initializer. If that initializer runs while a trace is active (e.g. the first scheduled
+   * delay/timeout under a span), the executor instrumentation captures a continuation on those
+   * static singletons that is never cancelled, leaking the pending trace. Disable async propagation
+   * while the type initializer runs. (RxJava 2 has no equivalent class.)
+   */
+  private static final ElementMatcher<TypeDescription> RXJAVA3_DISABLED_TYPE_INITIALIZERS =
+      named("io.reactivex.rxjava3.internal.schedulers.AbstractDirectTask");
+
   @Override
   public boolean onlyMatchKnownTypes() {
     return false; // known type list is not complete, so always expand search to consider hierarchy
@@ -77,7 +87,8 @@ public String[] knownMatchingTypes() {
       "net.sf.ehcache.store.disk.DiskStorageFactory",
       "org.springframework.jms.listener.DefaultMessageListenerContainer",
       "org.apache.activemq.broker.TransactionBroker",
-      "com.mongodb.internal.connection.DefaultConnectionPool$AsyncWorkManager"
+      "com.mongodb.internal.connection.DefaultConnectionPool$AsyncWorkManager",
+      "io.reactivex.rxjava3.internal.schedulers.AbstractDirectTask"
     };
   }
 
@@ -88,7 +99,10 @@ public String hierarchyMarkerType() {
 
   @Override
   public ElementMatcher<TypeDescription> hierarchyMatcher() {
-    return RX_WORKERS.or(GRPC_MANAGED_CHANNEL).or(REACTOR_DISABLED_TYPE_INITIALIZERS);
+    return RX_WORKERS
+        .or(GRPC_MANAGED_CHANNEL)
+        .or(REACTOR_DISABLED_TYPE_INITIALIZERS)
+        .or(RXJAVA3_DISABLED_TYPE_INITIALIZERS);
   }
 
   @Override
@@ -172,6 +186,8 @@ public void methodAdvice(MethodTransformer transformer) {
         advice);
     transformer.applyAdvice(
         isTypeInitializer().and(isDeclaredBy(REACTOR_DISABLED_TYPE_INITIALIZERS)), advice);
+    transformer.applyAdvice(
+        isTypeInitializer().and(isDeclaredBy(RXJAVA3_DISABLED_TYPE_INITIALIZERS)), advice);
   }
 
   public static class DisableAsyncAdvice {

@@ -0,0 +1,36 @@
+muzzle {
+  pass {
+    group = "io.reactivex.rxjava3"
+    module = "rxjava"
+    versions = "[3.0.0,)"
+  }
+  // Assert the rxjava3 advice never resolves against rxjava2 — the two namespaces
+  // must not overlap. rxjava3 references io.reactivex.rxjava3.core.*, absent from the
+  // rxjava2 artifact, so muzzle must fail to match it.
+  fail {
+    name = "rxjava2-must-not-match"
+    group = "io.reactivex.rxjava2"
+    module = "rxjava"
+    versions = "[2.0.0,)"
+  }
+}
+
+apply from: "$rootDir/gradle/java.gradle"
+
+addTestSuiteForDir('latestDepTest', 'test')
+
+dependencies {
+  compileOnly group: 'org.reactivestreams', name: 'reactive-streams', version: '1.0.3'
+  compileOnly group: 'io.reactivex.rxjava3', name: 'rxjava', version: '3.0.0'
+
+  testImplementation project(':dd-java-agent:instrumentation:datadog:tracing:trace-annotation')
+  testImplementation project(':dd-java-agent:instrumentation:opentelemetry:opentelemetry-annotations-1.20')
+  testImplementation group: 'io.reactivex.rxjava3', name: 'rxjava', version: '3.0.0'
+  testImplementation group: 'io.opentelemetry.instrumentation', name: 'opentelemetry-instrumentation-annotations', version: '1.28.0'
+
+  // Load the rxjava2 instrumenter at test runtime to prove the two versions coexist on
+  // the agent without interference (it stays dormant with only rxjava3 on the classpath).
+  testRuntimeOnly project(':dd-java-agent:instrumentation:rxjava:rxjava-2.0')
+
+  latestDepTestImplementation group: 'io.reactivex.rxjava3', name: 'rxjava', version: '+'
+}
@@ -0,0 +1,73 @@
+package datadog.trace.instrumentation.rxjava3;
+
+import static datadog.trace.agent.tooling.bytebuddy.matcher.NameMatchers.named;
+import static net.bytebuddy.matcher.ElementMatchers.isConstructor;
+import static net.bytebuddy.matcher.ElementMatchers.isMethod;
+import static net.bytebuddy.matcher.ElementMatchers.takesArgument;
+import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
+
+import datadog.context.Context;
+import datadog.context.ContextScope;
+import datadog.trace.agent.tooling.Instrumenter;
+import datadog.trace.bootstrap.InstrumentationContext;
+import datadog.trace.bootstrap.instrumentation.api.Java8BytecodeBridge;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.CompletableObserver;
+import net.bytebuddy.asm.Advice;
+
+public final class CompletableInstrumentation
+    implements Instrumenter.ForSingleType, Instrumenter.HasMethodAdvice {
+
+  @Override
+  public String instrumentedType() {
+    return "io.reactivex.rxjava3.core.Completable";
+  }
+
+  @Override
+  public void methodAdvice(MethodTransformer transformer) {
+    transformer.applyAdvice(isConstructor(), getClass().getName() + "$CaptureParentSpanAdvice");
+    transformer.applyAdvice(
+        isMethod()
+            .and(named("subscribe"))
+            .and(takesArguments(1))
+            .and(takesArgument(0, named("io.reactivex.rxjava3.core.CompletableObserver"))),
+        getClass().getName() + "$PropagateParentSpanAdvice");
+  }
+
+  public static class CaptureParentSpanAdvice {
+    @Advice.OnMethodExit(suppress = Throwable.class)
+    public static void onConstruct(@Advice.This final Completable completable) {
+      Context parentContext = Java8BytecodeBridge.getCurrentContext();
+      if (parentContext != null && parentContext != Java8BytecodeBridge.getRootContext()) {
+        InstrumentationContext.get(Completable.class, Context.class)
+            .put(completable, parentContext);
+      }
+    }
+  }
+
+  public static class PropagateParentSpanAdvice {
+    @Advice.OnMethodEnter(suppress = Throwable.class)
+    public static ContextScope onSubscribe(
+        @Advice.This final Completable completable,
+        @Advice.Argument(value = 0, readOnly = false) CompletableObserver observer) {
+      if (observer != null) {
+        Context parentContext =
+            InstrumentationContext.get(Completable.class, Context.class).get(completable);
+        if (parentContext != null) {
+          // wrap the observer so spans from its events treat the captured span as their parent
+          observer = new TracingCompletableObserver(observer, parentContext);
+          // attach the context here in case additional observers are created during subscribe
+          return parentContext.attach();
+        }
+      }
+      return null;
+    }
+
+    @Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class)
+    public static void closeScope(@Advice.Enter final ContextScope scope) {
+      if (scope != null) {
+        scope.close();
+      }
+    }
+  }
+}
@@ -0,0 +1,72 @@
+package datadog.trace.instrumentation.rxjava3;
+
+import static datadog.trace.agent.tooling.bytebuddy.matcher.NameMatchers.named;
+import static net.bytebuddy.matcher.ElementMatchers.isConstructor;
+import static net.bytebuddy.matcher.ElementMatchers.isMethod;
+import static net.bytebuddy.matcher.ElementMatchers.takesArgument;
+import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
+
+import datadog.context.Context;
+import datadog.context.ContextScope;
+import datadog.trace.agent.tooling.Instrumenter;
+import datadog.trace.bootstrap.InstrumentationContext;
+import datadog.trace.bootstrap.instrumentation.api.Java8BytecodeBridge;
+import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.core.FlowableSubscriber;
+import net.bytebuddy.asm.Advice;
+
+public final class FlowableInstrumentation
+    implements Instrumenter.ForSingleType, Instrumenter.HasMethodAdvice {
+
+  @Override
+  public String instrumentedType() {
+    return "io.reactivex.rxjava3.core.Flowable";
+  }
+
+  @Override
+  public void methodAdvice(MethodTransformer transformer) {
+    transformer.applyAdvice(isConstructor(), getClass().getName() + "$CaptureParentSpanAdvice");
+    transformer.applyAdvice(
+        isMethod()
+            .and(named("subscribe"))
+            .and(takesArguments(1))
+            .and(takesArgument(0, named("io.reactivex.rxjava3.core.FlowableSubscriber"))),
+        getClass().getName() + "$PropagateParentSpanAdvice");
+  }
+
+  public static class CaptureParentSpanAdvice {
+    @Advice.OnMethodExit(suppress = Throwable.class)
+    public static void onConstruct(@Advice.This final Flowable<?> flowable) {
+      Context parentContext = Java8BytecodeBridge.getCurrentContext();
+      if (parentContext != null && parentContext != Java8BytecodeBridge.getRootContext()) {
+        InstrumentationContext.get(Flowable.class, Context.class).put(flowable, parentContext);
+      }
+    }
+  }
+
+  public static class PropagateParentSpanAdvice {
+    @Advice.OnMethodEnter(suppress = Throwable.class)
+    public static ContextScope onSubscribe(
+        @Advice.This final Flowable<?> flowable,
+        @Advice.Argument(value = 0, readOnly = false) FlowableSubscriber<?> subscriber) {
+      if (subscriber != null) {
+        Context parentContext =
+            InstrumentationContext.get(Flowable.class, Context.class).get(flowable);
+        if (parentContext != null) {
+          // wrap the subscriber so spans from its events treat the captured span as their parent
+          subscriber = new TracingSubscriber<>(subscriber, parentContext);
+          // attach the context here in case additional subscribers are created during subscribe
+          return parentContext.attach();
+        }
+      }
+      return null;
+    }
+
+    @Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class)
+    public static void closeScope(@Advice.Enter final ContextScope scope) {
+      if (scope != null) {
+        scope.close();
+      }
+    }
+  }
+}
@@ -0,0 +1,70 @@
+package datadog.trace.instrumentation.rxjava3;
+
+import static datadog.trace.agent.tooling.bytebuddy.matcher.NameMatchers.named;
+import static net.bytebuddy.matcher.ElementMatchers.isConstructor;
+import static net.bytebuddy.matcher.ElementMatchers.isMethod;
+import static net.bytebuddy.matcher.ElementMatchers.takesArgument;
+import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
+
+import datadog.context.Context;
+import datadog.context.ContextScope;
+import datadog.trace.agent.tooling.Instrumenter;
+import datadog.trace.bootstrap.InstrumentationContext;
+import datadog.trace.bootstrap.instrumentation.api.Java8BytecodeBridge;
+import io.reactivex.rxjava3.core.Maybe;
+import io.reactivex.rxjava3.core.MaybeObserver;
+import net.bytebuddy.asm.Advice;
+
+public final class MaybeInstrumentation
+    implements Instrumenter.ForSingleType, Instrumenter.HasMethodAdvice {
+  @Override
+  public String instrumentedType() {
+    return "io.reactivex.rxjava3.core.Maybe";
+  }
+
+  @Override
+  public void methodAdvice(MethodTransformer transformer) {
+    transformer.applyAdvice(isConstructor(), getClass().getName() + "$CaptureParentSpanAdvice");
+    transformer.applyAdvice(
+        isMethod()
+            .and(named("subscribe"))
+            .and(takesArguments(1))
+            .and(takesArgument(0, named("io.reactivex.rxjava3.core.MaybeObserver"))),
+        getClass().getName() + "$PropagateParentSpanAdvice");
+  }
+
+  public static class CaptureParentSpanAdvice {
+    @Advice.OnMethodExit(suppress = Throwable.class)
+    public static void onConstruct(@Advice.This final Maybe<?> maybe) {
+      Context parentContext = Java8BytecodeBridge.getCurrentContext();
+      if (parentContext != null && parentContext != Java8BytecodeBridge.getRootContext()) {
+        InstrumentationContext.get(Maybe.class, Context.class).put(maybe, parentContext);
+      }
+    }
+  }
+
+  public static class PropagateParentSpanAdvice {
+    @Advice.OnMethodEnter(suppress = Throwable.class)
+    public static ContextScope onSubscribe(
+        @Advice.This final Maybe<?> maybe,
+        @Advice.Argument(value = 0, readOnly = false) MaybeObserver<?> observer) {
+      if (observer != null) {
+        Context parentContext = InstrumentationContext.get(Maybe.class, Context.class).get(maybe);
+        if (parentContext != null) {
+          // wrap the observer so spans from its events treat the captured span as their parent
+          observer = new TracingMaybeObserver<>(observer, parentContext);
+          // attach the context here in case additional observers are created during subscribe
+          return parentContext.attach();
+        }
+      }
+      return null;
+    }
+
+    @Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class)
+    public static void closeScope(@Advice.Enter final ContextScope scope) {
+      if (scope != null) {
+        scope.close();
+      }
+    }
+  }
+}
@@ -0,0 +1,71 @@
+package datadog.trace.instrumentation.rxjava3;
+
+import static datadog.trace.agent.tooling.bytebuddy.matcher.NameMatchers.named;
+import static net.bytebuddy.matcher.ElementMatchers.isConstructor;
+import static net.bytebuddy.matcher.ElementMatchers.isMethod;
+import static net.bytebuddy.matcher.ElementMatchers.takesArgument;
+import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
+
+import datadog.context.Context;
+import datadog.context.ContextScope;
+import datadog.trace.agent.tooling.Instrumenter;
+import datadog.trace.bootstrap.InstrumentationContext;
+import datadog.trace.bootstrap.instrumentation.api.Java8BytecodeBridge;
+import io.reactivex.rxjava3.core.Observable;
+import io.reactivex.rxjava3.core.Observer;
+import net.bytebuddy.asm.Advice;
+
+public final class ObservableInstrumentation
+    implements Instrumenter.ForSingleType, Instrumenter.HasMethodAdvice {
+  @Override
+  public String instrumentedType() {
+    return "io.reactivex.rxjava3.core.Observable";
+  }
+
+  @Override
+  public void methodAdvice(MethodTransformer transformer) {
+    transformer.applyAdvice(isConstructor(), getClass().getName() + "$CaptureParentSpanAdvice");
+    transformer.applyAdvice(
+        isMethod()
+            .and(named("subscribe"))
+            .and(takesArguments(1))
+            .and(takesArgument(0, named("io.reactivex.rxjava3.core.Observer"))),
+        getClass().getName() + "$PropagateParentSpanAdvice");
+  }
+
+  public static class CaptureParentSpanAdvice {
+    @Advice.OnMethodExit(suppress = Throwable.class)
+    public static void onConstruct(@Advice.This final Observable<?> observable) {
+      Context parentContext = Java8BytecodeBridge.getCurrentContext();
+      if (parentContext != null) {
+        InstrumentationContext.get(Observable.class, Context.class).put(observable, parentContext);
+      }
+    }
+  }
+
+  public static class PropagateParentSpanAdvice {
+    @Advice.OnMethodEnter(suppress = Throwable.class)
+    public static ContextScope onSubscribe(
+        @Advice.This final Observable<?> observable,
+        @Advice.Argument(value = 0, readOnly = false) Observer<?> observer) {
+      if (observer != null) {
+        Context parentContext =
+            InstrumentationContext.get(Observable.class, Context.class).get(observable);
+        if (parentContext != null && parentContext != Java8BytecodeBridge.getRootContext()) {
+          // wrap the observer so spans from its events treat the captured span as their parent
+          observer = new TracingObserver<>(observer, parentContext);
+          // attach the context here in case additional observers are created during subscribe
+          return parentContext.attach();
+        }
+      }
+      return null;
+    }
+
+    @Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class)
+    public static void closeScope(@Advice.Enter final ContextScope scope) {
+      if (scope != null) {
+        scope.close();
+      }
+    }
+  }
+}