Migrate executors test to java (#5596)

conventions/src/main/kotlin/io.opentelemetry.instrumentation.javaagent-testing.gradle.kts

@@ -39,8 +39,14 @@ dependencies {
 
   // Used by byte-buddy but not brought in as a transitive dependency
   compileOnly("com.google.code.findbugs:annotations")
+}
 
-  testImplementation("io.opentelemetry.javaagent:opentelemetry-testing-common")
+testing {
+  suites.withType(JvmTestSuite::class).configureEach {
+    dependencies {
+      implementation("io.opentelemetry.javaagent:opentelemetry-testing-common")
+    }
+  }
 }
 
 val testInstrumentation by configurations.creating {

conventions/src/main/kotlin/otel.java-conventions.gradle.kts

@@ -125,36 +125,46 @@ dependencies {
 
   compileOnly("com.google.code.findbugs:jsr305")
 
-  testImplementation("org.junit.jupiter:junit-jupiter-api")
-  testImplementation("org.junit.jupiter:junit-jupiter-params")
-  testRuntimeOnly("org.junit.jupiter:junit-jupiter-engine")
-  testRuntimeOnly("org.junit.vintage:junit-vintage-engine")
-
-  testImplementation("org.objenesis:objenesis")
-  testImplementation("org.spockframework:spock-core") {
-    // exclude optional dependencies
-    exclude(group = "cglib", module = "cglib-nodep")
-    exclude(group = "net.bytebuddy", module = "byte-buddy")
-    exclude(group = "org.junit.platform", module = "junit-platform-testkit")
-    exclude(group = "org.jetbrains", module = "annotations")
-    exclude(group = "org.objenesis", module = "objenesis")
-    exclude(group = "org.ow2.asm", module = "asm")
-  }
-  testImplementation("org.spockframework:spock-junit4") {
-    // spock-core is already added as dependency
-    // exclude it here to avoid pulling in optional dependencies
-    exclude(group = "org.spockframework", module = "spock-core")
-  }
-  testImplementation("ch.qos.logback:logback-classic")
-  testImplementation("org.slf4j:log4j-over-slf4j")
-  testImplementation("org.slf4j:jcl-over-slf4j")
-  testImplementation("org.slf4j:jul-to-slf4j")
-  testImplementation("com.github.stefanbirkner:system-rules")
-
   codenarc("org.codenarc:CodeNarc:2.2.0")
   codenarc(platform("org.codehaus.groovy:groovy-bom:3.0.9"))
 }
 
+testing {
+  suites.withType(JvmTestSuite::class).configureEach {
+    dependencies {
+      implementation("org.junit.jupiter:junit-jupiter-api")
+      implementation("org.junit.jupiter:junit-jupiter-params")
+      runtimeOnly("org.junit.jupiter:junit-jupiter-engine")
+      runtimeOnly("org.junit.vintage:junit-vintage-engine")
+
+      implementation("org.objenesis:objenesis")
+      implementation("org.spockframework:spock-core") {
+        with (this as ExternalDependency) {
+          // exclude optional dependencies
+          exclude(group = "cglib", module = "cglib-nodep")
+          exclude(group = "net.bytebuddy", module = "byte-buddy")
+          exclude(group = "org.junit.platform", module = "junit-platform-testkit")
+          exclude(group = "org.jetbrains", module = "annotations")
+          exclude(group = "org.objenesis", module = "objenesis")
+          exclude(group = "org.ow2.asm", module = "asm")
+        }
+      }
+      implementation("org.spockframework:spock-junit4") {
+        with (this as ExternalDependency) {
+          // spock-core is already added as dependency
+          // exclude it here to avoid pulling in optional dependencies
+          exclude(group = "org.spockframework", module = "spock-core")
+        }
+      }
+      implementation("ch.qos.logback:logback-classic")
+      implementation("org.slf4j:log4j-over-slf4j")
+      implementation("org.slf4j:jcl-over-slf4j")
+      implementation("org.slf4j:jul-to-slf4j")
+      implementation("com.github.stefanbirkner:system-rules")
+    }
+  }
+}
+
 tasks {
   named<Jar>("jar") {
     // By default Gradle Jar task can put multiple files with the same name

instrumentation/executors/javaagent/build.gradle.kts

@@ -8,10 +8,34 @@ muzzle {
   }
 }
 
-tasks.withType<Test>().configureEach {
-  jvmArgs("-Dotel.instrumentation.executors.include=ExecutorInstrumentationTest\$CustomThreadPoolExecutor")
-  jvmArgs("-Djava.awt.headless=true")
-  // ExecutorInstrumentationTest tess internal JDK class instrumentation
-  jvmArgs("--add-opens=java.base/java.util.concurrent=ALL-UNNAMED")
-  jvmArgs("-XX:+IgnoreUnrecognizedVMOptions")
+testing {
+  suites {
+    // CompletableFuture behaves differently if ForkJoinPool has no parallelism
+    val testNoParallelism by registering(JvmTestSuite::class) {
+      sources {
+        java {
+          setSrcDirs(listOf("src/test/java"))
+        }
+      }
+
+      targets {
+        all {
+          testTask.configure {
+            systemProperty("java.util.concurrent.ForkJoinPool.common.parallelism", 1)
+          }
+        }
+      }
+    }
+  }
+}
+
+tasks {
+  withType<Test>().configureEach {
+    jvmArgs("-Dotel.instrumentation.executors.include=io.opentelemetry.javaagent.instrumentation.javaconcurrent.ExecutorInstrumentationTest\$CustomThreadPoolExecutor")
+    jvmArgs("-Djava.awt.headless=true")
+  }
+
+  check {
+    dependsOn(testing.suites)
+  }
 }

instrumentation/executors/javaagent/src/test/groovy/CompletableFutureTest.groovy

@@ -1,302 +0,0 @@
-/*
- * Copyright The OpenTelemetry Authors
- * SPDX-License-Identifier: Apache-2.0
- */
-
-import io.opentelemetry.api.trace.SpanKind
-import io.opentelemetry.instrumentation.test.AgentInstrumentationSpecification
-import spock.lang.Requires
-
-import java.util.concurrent.ArrayBlockingQueue
-import java.util.concurrent.CompletableFuture
-import java.util.concurrent.ThreadPoolExecutor
-import java.util.concurrent.TimeUnit
-import java.util.function.Function
-import java.util.function.Supplier
-
-@Requires({ javaVersion >= 1.8 })
-class CompletableFutureTest extends AgentInstrumentationSpecification {
-
-  def "CompletableFuture test"() {
-    setup:
-    def pool = new ThreadPoolExecutor(1, 1, 1000, TimeUnit.NANOSECONDS, new ArrayBlockingQueue<Runnable>(1))
-    def differentPool = new ThreadPoolExecutor(1, 1, 1000, TimeUnit.NANOSECONDS, new ArrayBlockingQueue<Runnable>(1))
-    def supplier = new Supplier<String>() {
-      @Override
-      String get() {
-        runWithSpan("supplier") {}
-        sleep(1000)
-        return "a"
-      }
-    }
-
-    def function = new Function<String, String>() {
-      @Override
-      String apply(String s) {
-        runWithSpan("function") {}
-        return s + "c"
-      }
-    }
-
-    def result = new Supplier<String>() {
-      @Override
-      String get() {
-        runWithSpan("parent") {
-          return CompletableFuture.supplyAsync(supplier, pool)
-            .thenCompose({ s -> CompletableFuture.supplyAsync(new AppendingSupplier(s), differentPool) })
-            .thenApply(function)
-            .get()
-        }
-      }
-    }.get()
-
-    expect:
-    result == "abc"
-
-    assertTraces(1) {
-      trace(0, 4) {
-        span(0) {
-          name "parent"
-          kind SpanKind.INTERNAL
-          hasNoParent()
-        }
-        span(1) {
-          name "supplier"
-          kind SpanKind.INTERNAL
-          childOf span(0)
-        }
-        span(2) {
-          name "appendingSupplier"
-          kind SpanKind.INTERNAL
-          childOf span(0)
-        }
-        span(3) {
-          name "function"
-          kind SpanKind.INTERNAL
-          childOf span(0)
-        }
-      }
-    }
-
-    cleanup:
-    pool?.shutdown()
-    differentPool?.shutdown()
-  }
-
-  def "test supplyAsync"() {
-    when:
-    CompletableFuture<String> completableFuture = runWithSpan("parent") {
-      def result = CompletableFuture.supplyAsync {
-        runWithSpan("child") {
-          "done"
-        }
-      }
-      return result
-    }
-
-    then:
-    completableFuture.get() == "done"
-
-    and:
-    assertTraces(1) {
-      trace(0, 2) {
-        span(0) {
-          name "parent"
-          kind SpanKind.INTERNAL
-          hasNoParent()
-        }
-        span(1) {
-          name "child"
-          kind SpanKind.INTERNAL
-          childOf span(0)
-        }
-      }
-    }
-  }
-
-  def "test thenApply"() {
-    when:
-    CompletableFuture<String> completableFuture = runWithSpan("parent") {
-      CompletableFuture.supplyAsync {
-        "done"
-      }.thenApply { result ->
-        runWithSpan("child") {
-          result
-        }
-      }
-    }
-
-    then:
-    completableFuture.get() == "done"
-
-    and:
-    assertTraces(1) {
-      trace(0, 2) {
-        span(0) {
-          name "parent"
-          kind SpanKind.INTERNAL
-          hasNoParent()
-        }
-        span(1) {
-          name "child"
-          kind SpanKind.INTERNAL
-          childOf span(0)
-        }
-      }
-    }
-  }
-
-  def "test thenApplyAsync"() {
-    when:
-    CompletableFuture<String> completableFuture = runWithSpan("parent") {
-      def result = CompletableFuture.supplyAsync {
-        "done"
-      }.thenApplyAsync { result ->
-        runWithSpan("child") {
-          result
-        }
-      }
-      return result
-    }
-
-    then:
-    completableFuture.get() == "done"
-
-    and:
-    assertTraces(1) {
-      trace(0, 2) {
-        span(0) {
-          name "parent"
-          kind SpanKind.INTERNAL
-          hasNoParent()
-        }
-        span(1) {
-          name "child"
-          kind SpanKind.INTERNAL
-          childOf span(0)
-        }
-      }
-    }
-  }
-
-  def "test thenCompose"() {
-    when:
-    CompletableFuture<String> completableFuture = runWithSpan("parent") {
-      def result = CompletableFuture.supplyAsync {
-        "done"
-      }.thenCompose { result ->
-        CompletableFuture.supplyAsync {
-          runWithSpan("child") {
-            result
-          }
-        }
-      }
-      return result
-    }
-
-    then:
-    completableFuture.get() == "done"
-
-    and:
-    assertTraces(1) {
-      trace(0, 2) {
-        span(0) {
-          name "parent"
-          kind SpanKind.INTERNAL
-          hasNoParent()
-        }
-        span(1) {
-          name "child"
-          kind SpanKind.INTERNAL
-          childOf span(0)
-        }
-      }
-    }
-  }
-
-  def "test thenComposeAsync"() {
-    when:
-    CompletableFuture<String> completableFuture = runWithSpan("parent") {
-      def result = CompletableFuture.supplyAsync {
-        "done"
-      }.thenComposeAsync { result ->
-        CompletableFuture.supplyAsync {
-          runWithSpan("child") {
-            result
-          }
-        }
-      }
-      return result
-    }
-
-    then:
-    completableFuture.get() == "done"
-
-    and:
-    assertTraces(1) {
-      trace(0, 2) {
-        span(0) {
-          name "parent"
-          kind SpanKind.INTERNAL
-          hasNoParent()
-        }
-        span(1) {
-          name "child"
-          kind SpanKind.INTERNAL
-          childOf span(0)
-        }
-      }
-    }
-  }
-
-  def "test compose and apply"() {
-    when:
-    CompletableFuture<String> completableFuture = runWithSpan("parent") {
-      def result = CompletableFuture.supplyAsync {
-        "do"
-      }.thenCompose { result ->
-        CompletableFuture.supplyAsync {
-          result + "ne"
-        }
-      }.thenApplyAsync { result ->
-        runWithSpan("child") {
-          result
-        }
-      }
-      return result
-    }
-
-    then:
-    completableFuture.get() == "done"
-
-    and:
-    assertTraces(1) {
-      trace(0, 2) {
-        span(0) {
-          name "parent"
-          kind SpanKind.INTERNAL
-          hasNoParent()
-        }
-        span(1) {
-          name "child"
-          kind SpanKind.INTERNAL
-          childOf span(0)
-        }
-      }
-    }
-  }
-
-  class AppendingSupplier implements Supplier<String> {
-    String letter
-
-    AppendingSupplier(String letter) {
-      this.letter = letter
-    }
-
-    @Override
-    String get() {
-      runWithSpan("appendingSupplier") {}
-      return letter + "b"
-    }
-  }
-}

instrumentation/executors/javaagent/src/test/groovy/ExecutorInstrumentationTest.groovy

@@ -1,359 +0,0 @@
-/*
- * Copyright The OpenTelemetry Authors
- * SPDX-License-Identifier: Apache-2.0
- */
-
-import io.opentelemetry.api.trace.SpanKind
-import io.opentelemetry.instrumentation.test.AgentInstrumentationSpecification
-import spock.lang.Shared
-
-import java.lang.reflect.InvocationTargetException
-import java.util.concurrent.AbstractExecutorService
-import java.util.concurrent.ArrayBlockingQueue
-import java.util.concurrent.Callable
-import java.util.concurrent.CompletableFuture
-import java.util.concurrent.ExecutionException
-import java.util.concurrent.ExecutorService
-import java.util.concurrent.ForkJoinPool
-import java.util.concurrent.ForkJoinTask
-import java.util.concurrent.Future
-import java.util.concurrent.LinkedBlockingQueue
-import java.util.concurrent.RejectedExecutionException
-import java.util.concurrent.ScheduledThreadPoolExecutor
-import java.util.concurrent.ThreadPoolExecutor
-import java.util.concurrent.TimeUnit
-import java.util.concurrent.TimeoutException
-
-class ExecutorInstrumentationTest extends AgentInstrumentationSpecification {
-
-  @Shared
-  def executeRunnable = { e, c -> e.execute((Runnable) c) }
-  @Shared
-  def executeForkJoinTask = { e, c -> e.execute((ForkJoinTask) c) }
-  @Shared
-  def submitRunnable = { e, c -> e.submit((Runnable) c) }
-  @Shared
-  def submitCallable = { e, c -> e.submit((Callable) c) }
-  @Shared
-  def submitForkJoinTask = { e, c -> e.submit((ForkJoinTask) c) }
-  @Shared
-  def invokeAll = { e, c -> e.invokeAll([(Callable) c]) }
-  @Shared
-  def invokeAllTimeout = { e, c -> e.invokeAll([(Callable) c], 10, TimeUnit.SECONDS) }
-  @Shared
-  def invokeAny = { e, c -> e.invokeAny([(Callable) c]) }
-  @Shared
-  def invokeAnyTimeout = { e, c -> e.invokeAny([(Callable) c], 10, TimeUnit.SECONDS) }
-  @Shared
-  def invokeForkJoinTask = { e, c -> e.invoke((ForkJoinTask) c) }
-  @Shared
-  def scheduleRunnable = { e, c -> e.schedule((Runnable) c, 10, TimeUnit.MILLISECONDS) }
-  @Shared
-  def scheduleCallable = { e, c -> e.schedule((Callable) c, 10, TimeUnit.MILLISECONDS) }
-
-  def "#poolName '#testName' propagates"() {
-    setup:
-    def pool = poolImpl
-    def m = method
-
-    new Runnable() {
-      @Override
-      void run() {
-        runWithSpan("parent") {
-          // this child will have a span
-          def child1 = new JavaAsyncChild()
-          // this child won't
-          def child2 = new JavaAsyncChild(false, false)
-          m(pool, child1)
-          m(pool, child2)
-          child1.waitForCompletion()
-          child2.waitForCompletion()
-        }
-      }
-    }.run()
-
-    expect:
-    assertTraces(1) {
-      trace(0, 2) {
-        span(0) {
-          name "parent"
-          kind SpanKind.INTERNAL
-          hasNoParent()
-        }
-        span(1) {
-          name "asyncChild"
-          kind SpanKind.INTERNAL
-          childOf span(0)
-        }
-      }
-    }
-
-    cleanup:
-    if (pool.hasProperty("shutdown")) {
-      pool.shutdown()
-      pool.awaitTermination(10, TimeUnit.SECONDS)
-    }
-
-    // Unfortunately, there's no simple way to test the cross product of methods/pools.
-    where:
-    testName                 | method              | poolImpl
-    "execute Runnable"       | executeRunnable     | new ThreadPoolExecutor(1, 1, 1000, TimeUnit.NANOSECONDS, new ArrayBlockingQueue<Runnable>(1))
-    "submit Runnable"        | submitRunnable      | new ThreadPoolExecutor(1, 1, 1000, TimeUnit.NANOSECONDS, new ArrayBlockingQueue<Runnable>(1))
-    "submit Callable"        | submitCallable      | new ThreadPoolExecutor(1, 1, 1000, TimeUnit.NANOSECONDS, new ArrayBlockingQueue<Runnable>(1))
-    "invokeAll"              | invokeAll           | new ThreadPoolExecutor(1, 1, 1000, TimeUnit.NANOSECONDS, new ArrayBlockingQueue<Runnable>(1))
-    "invokeAll with timeout" | invokeAllTimeout    | new ThreadPoolExecutor(1, 1, 1000, TimeUnit.NANOSECONDS, new ArrayBlockingQueue<Runnable>(1))
-    "invokeAny"              | invokeAny           | new ThreadPoolExecutor(1, 1, 1000, TimeUnit.NANOSECONDS, new ArrayBlockingQueue<Runnable>(1))
-    "invokeAny with timeout" | invokeAnyTimeout    | new ThreadPoolExecutor(1, 1, 1000, TimeUnit.NANOSECONDS, new ArrayBlockingQueue<Runnable>(1))
-
-    // Scheduled executor has additional methods and also may get disabled because it wraps tasks
-    "execute Runnable"       | executeRunnable     | new ScheduledThreadPoolExecutor(1)
-    "submit Runnable"        | submitRunnable      | new ScheduledThreadPoolExecutor(1)
-    "submit Callable"        | submitCallable      | new ScheduledThreadPoolExecutor(1)
-    "invokeAll"              | invokeAll           | new ScheduledThreadPoolExecutor(1)
-    "invokeAll with timeout" | invokeAllTimeout    | new ScheduledThreadPoolExecutor(1)
-    "invokeAny"              | invokeAny           | new ScheduledThreadPoolExecutor(1)
-    "invokeAny with timeout" | invokeAnyTimeout    | new ScheduledThreadPoolExecutor(1)
-    "schedule Runnable"      | scheduleRunnable    | new ScheduledThreadPoolExecutor(1)
-    "schedule Callable"      | scheduleCallable    | new ScheduledThreadPoolExecutor(1)
-
-    // ForkJoinPool has additional set of method overloads for ForkJoinTask to deal with
-    "execute Runnable"       | executeRunnable     | new ForkJoinPool()
-    "execute ForkJoinTask"   | executeForkJoinTask | new ForkJoinPool()
-    "submit Runnable"        | submitRunnable      | new ForkJoinPool()
-    "submit Callable"        | submitCallable      | new ForkJoinPool()
-    "submit ForkJoinTask"    | submitForkJoinTask  | new ForkJoinPool()
-    "invoke ForkJoinTask"    | invokeForkJoinTask  | new ForkJoinPool()
-    "invokeAll"              | invokeAll           | new ForkJoinPool()
-    "invokeAll with timeout" | invokeAllTimeout    | new ForkJoinPool()
-    "invokeAny"              | invokeAny           | new ForkJoinPool()
-    "invokeAny with timeout" | invokeAnyTimeout    | new ForkJoinPool()
-
-    // CustomThreadPoolExecutor would normally be disabled except enabled above.
-    "execute Runnable"       | executeRunnable     | new CustomThreadPoolExecutor()
-    "submit Runnable"        | submitRunnable      | new CustomThreadPoolExecutor()
-    "submit Callable"        | submitCallable      | new CustomThreadPoolExecutor()
-    "invokeAll"              | invokeAll           | new CustomThreadPoolExecutor()
-    "invokeAll with timeout" | invokeAllTimeout    | new CustomThreadPoolExecutor()
-    "invokeAny"              | invokeAny           | new CustomThreadPoolExecutor()
-    "invokeAny with timeout" | invokeAnyTimeout    | new CustomThreadPoolExecutor()
-
-    // Internal executor used by CompletableFuture
-    "execute Runnable"       | executeRunnable     | new CompletableFuture.ThreadPerTaskExecutor()
-    poolName = poolImpl.class.simpleName
-  }
-
-  def "#poolName '#testName' wrap lambdas"() {
-    setup:
-    ExecutorService pool = poolImpl
-    def m = method
-    def w = wrap
-
-    JavaAsyncChild child = new JavaAsyncChild(true, true)
-    new Runnable() {
-      @Override
-      void run() {
-        runWithSpan("parent") {
-          m(pool, w(child))
-        }
-      }
-    }.run()
-    // We block in child to make sure spans close in predictable order
-    child.unblock()
-    child.waitForCompletion()
-
-    expect:
-    assertTraces(1) {
-      trace(0, 2) {
-        span(0) {
-          name "parent"
-          kind SpanKind.INTERNAL
-          hasNoParent()
-        }
-        span(1) {
-          name "asyncChild"
-          kind SpanKind.INTERNAL
-          childOf span(0)
-        }
-      }
-    }
-
-    cleanup:
-    pool.shutdown()
-    pool.awaitTermination(10, TimeUnit.SECONDS)
-
-    where:
-    testName            | method           | wrap                           | poolImpl
-    "execute Runnable"  | executeRunnable  | { LambdaGen.wrapRunnable(it) } | new ScheduledThreadPoolExecutor(1)
-    "submit Runnable"   | submitRunnable   | { LambdaGen.wrapRunnable(it) } | new ScheduledThreadPoolExecutor(1)
-    "submit Callable"   | submitCallable   | { LambdaGen.wrapCallable(it) } | new ScheduledThreadPoolExecutor(1)
-    "schedule Runnable" | scheduleRunnable | { LambdaGen.wrapRunnable(it) } | new ScheduledThreadPoolExecutor(1)
-    "schedule Callable" | scheduleCallable | { LambdaGen.wrapCallable(it) } | new ScheduledThreadPoolExecutor(1)
-    poolName = poolImpl.class.simpleName
-  }
-
-  def "#poolName '#testName' reports after canceled jobs"() {
-    setup:
-    ExecutorService pool = poolImpl
-    def m = method
-    List<JavaAsyncChild> children = new ArrayList<>()
-    List<Future> jobFutures = new ArrayList<>()
-
-    new Runnable() {
-      @Override
-      void run() {
-        runWithSpan("parent") {
-          try {
-            for (int i = 0; i < 20; ++i) {
-              // Our current instrumentation instrumentation does not behave very well
-              // if we try to reuse Callable/Runnable. Namely we would be getting 'orphaned'
-              // child traces sometimes since state can contain only one parent span - and
-              // we do not really have a good way for attributing work to correct parent span
-              // if we reuse Callable/Runnable.
-              // Solution for now is to never reuse a Callable/Runnable.
-              JavaAsyncChild child = new JavaAsyncChild(false, true)
-              children.add(child)
-              try {
-                Future f = m(pool, child)
-                jobFutures.add(f)
-              } catch (InvocationTargetException e) {
-                throw e.getCause()
-              }
-            }
-          } catch (RejectedExecutionException e) {
-          }
-
-          for (Future f : jobFutures) {
-            f.cancel(false)
-          }
-          for (JavaAsyncChild child : children) {
-            child.unblock()
-          }
-        }
-      }
-    }.run()
-
-
-    expect:
-    waitForTraces(1).size() == 1
-
-    pool.shutdown()
-    pool.awaitTermination(10, TimeUnit.SECONDS)
-
-    where:
-    testName            | method           | poolImpl
-    "submit Runnable"   | submitRunnable   | new ThreadPoolExecutor(1, 1, 1000, TimeUnit.NANOSECONDS, new ArrayBlockingQueue<Runnable>(1))
-    "submit Callable"   | submitCallable   | new ThreadPoolExecutor(1, 1, 1000, TimeUnit.NANOSECONDS, new ArrayBlockingQueue<Runnable>(1))
-
-    // Scheduled executor has additional methods and also may get disabled because it wraps tasks
-    "submit Runnable"   | submitRunnable   | new ScheduledThreadPoolExecutor(1)
-    "submit Callable"   | submitCallable   | new ScheduledThreadPoolExecutor(1)
-    "schedule Runnable" | scheduleRunnable | new ScheduledThreadPoolExecutor(1)
-    "schedule Callable" | scheduleCallable | new ScheduledThreadPoolExecutor(1)
-
-    // ForkJoinPool has additional set of method overloads for ForkJoinTask to deal with
-    "submit Runnable"   | submitRunnable   | new ForkJoinPool()
-    "submit Callable"   | submitCallable   | new ForkJoinPool()
-    poolName = poolImpl.class.simpleName
-  }
-
-  static class CustomThreadPoolExecutor extends AbstractExecutorService {
-    volatile running = true
-    def workQueue = new LinkedBlockingQueue<Runnable>(10)
-
-    def worker = new Runnable() {
-      void run() {
-        try {
-          while (running) {
-            def runnable = workQueue.take()
-            runnable.run()
-          }
-        } catch (InterruptedException e) {
-          Thread.currentThread().interrupt()
-        } catch (Exception e) {
-          e.printStackTrace()
-        }
-      }
-    }
-
-    def workerThread = new Thread(worker, "ExecutorTestThread")
-
-    private CustomThreadPoolExecutor() {
-      workerThread.start()
-    }
-
-    @Override
-    void shutdown() {
-      running = false
-      workerThread.interrupt()
-    }
-
-    @Override
-    List<Runnable> shutdownNow() {
-      running = false
-      workerThread.interrupt()
-      return []
-    }
-
-    @Override
-    boolean isShutdown() {
-      return !running
-    }
-
-    @Override
-    boolean isTerminated() {
-      return workerThread.isAlive()
-    }
-
-    @Override
-    boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
-      workerThread.join(unit.toMillis(timeout))
-      return true
-    }
-
-    @Override
-    def <T> Future<T> submit(Callable<T> task) {
-      def future = newTaskFor(task)
-      execute(future)
-      return future
-    }
-
-    @Override
-    def <T> Future<T> submit(Runnable task, T result) {
-      def future = newTaskFor(task, result)
-      execute(future)
-      return future
-    }
-
-    @Override
-    Future<?> submit(Runnable task) {
-      def future = newTaskFor(task, null)
-      execute(future)
-      return future
-    }
-
-    @Override
-    def <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException {
-      return super.invokeAll(tasks)
-    }
-
-    @Override
-    def <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException {
-      return super.invokeAll(tasks)
-    }
-
-    @Override
-    def <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException {
-      return super.invokeAny(tasks)
-    }
-
-    @Override
-    def <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
-      return super.invokeAny(tasks)
-    }
-
-    @Override
-    void execute(Runnable command) {
-      workQueue.put(command)
-    }
-  }
-}

instrumentation/executors/javaagent/src/test/groovy/ForkJoinTaskTest.groovy

@@ -1,34 +0,0 @@
-/*
- * Copyright The OpenTelemetry Authors
- * SPDX-License-Identifier: Apache-2.0
- */
-
-import io.opentelemetry.instrumentation.test.AgentInstrumentationSpecification
-
-import java.util.stream.IntStream
-
-class ForkJoinTaskTest extends AgentInstrumentationSpecification {
-
-  def "test parallel"() {
-    when:
-    runWithSpan("parent") {
-      IntStream.range(0, 20)
-        .parallel()
-        .forEach({ runWithSpan("child") {} })
-    }
-
-    then:
-    assertTraces(1) {
-      trace(0, 21) {
-        span(0) {
-          name "parent"
-        }
-        (1..20).each { index ->
-          span(index) {
-            childOf(span(0))
-          }
-        }
-      }
-    }
-  }
-}

instrumentation/executors/javaagent/src/test/groovy/ModuleInjectionTest.groovy

@@ -1,26 +0,0 @@
-/*
- * Copyright The OpenTelemetry Authors
- * SPDX-License-Identifier: Apache-2.0
- */
-
-import io.opentelemetry.instrumentation.test.AgentInstrumentationSpecification
-
-import javax.swing.*
-
-/**
- * This class tests that we correctly add module references when instrumenting
- */
-class ModuleInjectionTest extends AgentInstrumentationSpecification {
-  /**
-   * There's nothing special about RepaintManager other than
-   * it's in a module (java.desktop) that doesn't read the "unnamed module" and it
-   * creates an instrumented runnable in its constructor
-   */
-  def "test instrumenting java.desktop class"() {
-    when:
-    new RepaintManager()
-
-    then:
-    noExceptionThrown()
-  }
-}

instrumentation/executors/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/javaconcurrent/CompletableFutureTest.java

@@ -0,0 +1,214 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.javaconcurrent;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.instrumentation.testing.junit.AgentInstrumentationExtension;
+import io.opentelemetry.instrumentation.testing.junit.InstrumentationExtension;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.function.Supplier;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+class CompletableFutureTest {
+
+  @RegisterExtension
+  static final InstrumentationExtension testing = AgentInstrumentationExtension.create();
+
+  @Test
+  void multipleCallbacks() {
+    ExecutorService executor = Executors.newSingleThreadExecutor();
+    ExecutorService executor2 = Executors.newSingleThreadExecutor();
+
+    String result;
+    try {
+      result =
+          testing.runWithSpan(
+              "parent",
+              () ->
+                  CompletableFuture.supplyAsync(
+                          () -> {
+                            testing.runWithSpan("supplier", () -> {});
+                            try {
+                              Thread.sleep(1);
+                            } catch (InterruptedException e) {
+                              Thread.currentThread().interrupt();
+                              throw new AssertionError(e);
+                            }
+                            return "a";
+                          },
+                          executor)
+                      .thenCompose(
+                          s -> CompletableFuture.supplyAsync(new AppendingSupplier(s), executor2))
+                      .thenApply(
+                          s -> {
+                            testing.runWithSpan("function", () -> {});
+                            return s + "c";
+                          })
+                      .get());
+    } catch (Exception e) {
+      throw new AssertionError(e);
+    }
+
+    assertThat(result).isEqualTo("abc");
+
+    testing.waitAndAssertTraces(
+        trace ->
+            trace.hasSpansSatisfyingExactly(
+                span -> span.hasName("parent").hasKind(SpanKind.INTERNAL).hasNoParent(),
+                span ->
+                    span.hasName("supplier").hasKind(SpanKind.INTERNAL).hasParent(trace.getSpan(0)),
+                span ->
+                    span.hasName("appendingSupplier")
+                        .hasKind(SpanKind.INTERNAL)
+                        .hasParent(trace.getSpan(0)),
+                span ->
+                    span.hasName("function")
+                        .hasKind(SpanKind.INTERNAL)
+                        .hasParent(trace.getSpan(0))));
+
+    executor.shutdown();
+    executor2.shutdown();
+  }
+
+  @Test
+  void supplyAsync() {
+    CompletableFuture<String> future =
+        testing.runWithSpan(
+            "parent",
+            () -> CompletableFuture.supplyAsync(() -> testing.runWithSpan("child", () -> "done")));
+
+    assertThat(future.join()).isEqualTo("done");
+
+    testing.waitAndAssertTraces(
+        trace ->
+            trace.hasSpansSatisfyingExactly(
+                span -> span.hasName("parent").hasKind(SpanKind.INTERNAL).hasNoParent(),
+                span ->
+                    span.hasName("child").hasKind(SpanKind.INTERNAL).hasParent(trace.getSpan(0))));
+  }
+
+  @Test
+  void thenApply() {
+    CompletableFuture<String> future =
+        testing.runWithSpan(
+            "parent",
+            () ->
+                CompletableFuture.supplyAsync(() -> "done")
+                    .thenApply(result -> testing.runWithSpan("child", () -> result)));
+
+    assertThat(future.join()).isEqualTo("done");
+
+    testing.waitAndAssertTraces(
+        trace ->
+            trace.hasSpansSatisfyingExactly(
+                span -> span.hasName("parent").hasKind(SpanKind.INTERNAL).hasNoParent(),
+                span ->
+                    span.hasName("child").hasKind(SpanKind.INTERNAL).hasParent(trace.getSpan(0))));
+  }
+
+  @Test
+  void thenApplyAsync() {
+    CompletableFuture<String> future =
+        testing.runWithSpan(
+            "parent",
+            () ->
+                CompletableFuture.supplyAsync(() -> "done")
+                    .thenApplyAsync(result -> testing.runWithSpan("child", () -> result)));
+
+    assertThat(future.join()).isEqualTo("done");
+
+    testing.waitAndAssertTraces(
+        trace ->
+            trace.hasSpansSatisfyingExactly(
+                span -> span.hasName("parent").hasKind(SpanKind.INTERNAL).hasNoParent(),
+                span ->
+                    span.hasName("child").hasKind(SpanKind.INTERNAL).hasParent(trace.getSpan(0))));
+  }
+
+  @Test
+  void thenCompose() {
+    CompletableFuture<String> future =
+        testing.runWithSpan(
+            "parent",
+            () ->
+                CompletableFuture.supplyAsync(() -> "done")
+                    .thenCompose(
+                        result ->
+                            CompletableFuture.supplyAsync(
+                                () -> testing.runWithSpan("child", () -> result))));
+
+    assertThat(future.join()).isEqualTo("done");
+
+    testing.waitAndAssertTraces(
+        trace ->
+            trace.hasSpansSatisfyingExactly(
+                span -> span.hasName("parent").hasKind(SpanKind.INTERNAL).hasNoParent(),
+                span ->
+                    span.hasName("child").hasKind(SpanKind.INTERNAL).hasParent(trace.getSpan(0))));
+  }
+
+  @Test
+  void thenComposeAsync() {
+    CompletableFuture<String> future =
+        testing.runWithSpan(
+            "parent",
+            () ->
+                CompletableFuture.supplyAsync(() -> "done")
+                    .thenComposeAsync(
+                        result ->
+                            CompletableFuture.supplyAsync(
+                                () -> testing.runWithSpan("child", () -> result))));
+
+    assertThat(future.join()).isEqualTo("done");
+
+    testing.waitAndAssertTraces(
+        trace ->
+            trace.hasSpansSatisfyingExactly(
+                span -> span.hasName("parent").hasKind(SpanKind.INTERNAL).hasNoParent(),
+                span ->
+                    span.hasName("child").hasKind(SpanKind.INTERNAL).hasParent(trace.getSpan(0))));
+  }
+
+  @Test
+  void thenComposeAndApply() {
+    CompletableFuture<String> future =
+        testing.runWithSpan(
+            "parent",
+            () ->
+                CompletableFuture.supplyAsync(() -> "do")
+                    .thenCompose(result -> CompletableFuture.supplyAsync(() -> result + "ne"))
+                    .thenApplyAsync(result -> testing.runWithSpan("child", () -> result)));
+
+    assertThat(future.join()).isEqualTo("done");
+
+    testing.waitAndAssertTraces(
+        trace ->
+            trace.hasSpansSatisfyingExactly(
+                span -> span.hasName("parent").hasKind(SpanKind.INTERNAL).hasNoParent(),
+                span ->
+                    span.hasName("child").hasKind(SpanKind.INTERNAL).hasParent(trace.getSpan(0))));
+  }
+
+  static final class AppendingSupplier implements Supplier<String> {
+
+    private final String letter;
+
+    AppendingSupplier(String letter) {
+      this.letter = letter;
+    }
+
+    @Override
+    public String get() {
+      testing.runWithSpan("appendingSupplier", () -> {});
+      return letter + "b";
+    }
+  }
+}

instrumentation/executors/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/javaconcurrent/ExecutorInstrumentationTest.java

@@ -0,0 +1,375 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.javaconcurrent;
+
+import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.instrumentation.testing.junit.AgentInstrumentationExtension;
+import io.opentelemetry.instrumentation.testing.junit.InstrumentationExtension;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.AbstractExecutorService;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.ForkJoinTask;
+import java.util.concurrent.Future;
+import java.util.concurrent.RunnableFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.function.ThrowingConsumer;
+
+@SuppressWarnings("ClassCanBeStatic")
+class ExecutorInstrumentationTest {
+
+  @RegisterExtension
+  static final InstrumentationExtension testing = AgentInstrumentationExtension.create();
+
+  @Nested
+  @TestInstance(TestInstance.Lifecycle.PER_CLASS)
+  class ThreadPoolExecutorTest extends AbstractExecutorServiceTest<ThreadPoolExecutor> {
+    ThreadPoolExecutorTest() {
+      super(new ThreadPoolExecutor(1, 1, 1000, TimeUnit.NANOSECONDS, new ArrayBlockingQueue<>(20)));
+    }
+  }
+
+  @Nested
+  @TestInstance(TestInstance.Lifecycle.PER_CLASS)
+  class ScheduledThreadPoolExecutorTest
+      extends AbstractExecutorServiceTest<ScheduledThreadPoolExecutor> {
+    ScheduledThreadPoolExecutorTest() {
+      super(new ScheduledThreadPoolExecutor(1));
+    }
+
+    @Test
+    void scheduleRunnable() {
+      executeTwoTasks(task -> executor().schedule((Runnable) task, 10, TimeUnit.MICROSECONDS));
+    }
+
+    @Test
+    void scheduleCallable() {
+      executeTwoTasks(task -> executor().schedule((Callable<?>) task, 10, TimeUnit.MICROSECONDS));
+    }
+
+    @Test
+    void scheduleLambdaRunnable() {
+      executeTwoTasks(task -> executor().schedule(() -> task.run(), 10, TimeUnit.MICROSECONDS));
+    }
+
+    @Test
+    void scheduleLambdaCallable() {
+      executeTwoTasks(
+          task ->
+              executor()
+                  .schedule(
+                      () -> {
+                        task.run();
+                        return null;
+                      },
+                      10,
+                      TimeUnit.MICROSECONDS));
+    }
+
+    @Test
+    void scheduleRunnableAndCancel() {
+      executeAndCancelTasks(
+          task -> executor().schedule((Runnable) task, 10, TimeUnit.MICROSECONDS));
+    }
+
+    @Test
+    void scheduleCallableAndCancel() {
+      executeAndCancelTasks(
+          task -> executor().schedule((Callable<?>) task, 10, TimeUnit.MICROSECONDS));
+    }
+  }
+
+  @Nested
+  @TestInstance(TestInstance.Lifecycle.PER_CLASS)
+  class ForkJoinPoolTest extends AbstractExecutorServiceTest<ForkJoinPool> {
+    ForkJoinPoolTest() {
+      super(new ForkJoinPool(20));
+    }
+
+    @Test
+    void invokeForkJoinTask() {
+      executeTwoTasks(task -> executor().invoke((ForkJoinTask<?>) task));
+    }
+
+    @Test
+    void submitForkJoinTask() {
+      executeTwoTasks(task -> executor().submit((ForkJoinTask<?>) task));
+    }
+  }
+
+  // CustomThreadPoolExecutor would normally be disabled except enabled by system property.
+  @Nested
+  @TestInstance(TestInstance.Lifecycle.PER_CLASS)
+  class CustomThreadPoolExecutorTest extends AbstractExecutorServiceTest<CustomThreadPoolExecutor> {
+    CustomThreadPoolExecutorTest() {
+      super(new CustomThreadPoolExecutor());
+    }
+  }
+
+  abstract static class AbstractExecutorServiceTest<T extends ExecutorService> {
+    private final T executor;
+
+    AbstractExecutorServiceTest(T executor) {
+      this.executor = executor;
+    }
+
+    T executor() {
+      return executor;
+    }
+
+    @AfterAll
+    void shutdown() throws Exception {
+      executor.shutdown();
+      executor.awaitTermination(10, TimeUnit.SECONDS);
+    }
+
+    @Test
+    void executeRunnable() {
+      executeTwoTasks(executor::execute);
+    }
+
+    @Test
+    void submitRunnable() {
+      executeTwoTasks(task -> executor.submit((Runnable) task));
+    }
+
+    @Test
+    void submitCallable() {
+      executeTwoTasks(task -> executor.submit((Callable<?>) task));
+    }
+
+    @Test
+    void invokeAll() {
+      executeTwoTasks(task -> executor.invokeAll(Collections.singleton(task)));
+    }
+
+    @Test
+    void invokeAllWithTimeout() {
+      executeTwoTasks(
+          task -> executor.invokeAll(Collections.singleton(task), 10, TimeUnit.SECONDS));
+    }
+
+    @Test
+    void invokeAny() {
+      executeTwoTasks(task -> executor.invokeAny(Collections.singleton(task)));
+    }
+
+    @Test
+    void invokeAnyWithTimeout() {
+      executeTwoTasks(
+          task -> executor.invokeAny(Collections.singleton(task), 10, TimeUnit.SECONDS));
+    }
+
+    @Test
+    void executeLambdaRunnable() {
+      executeTwoTasks(task -> executor.execute(() -> task.run()));
+    }
+
+    @Test
+    void submitLambdaRunnable() {
+      executeTwoTasks(task -> executor.submit(() -> task.run()));
+    }
+
+    @Test
+    void submitLambdaCallable() {
+      executeTwoTasks(
+          task ->
+              executor.submit(
+                  () -> {
+                    task.run();
+                    return null;
+                  }));
+    }
+
+    @Test
+    void submitRunnableAndCancel() {
+      executeAndCancelTasks(task -> executor.submit((Runnable) task));
+    }
+
+    @Test
+    void submitCallableAndCancel() {
+      executeAndCancelTasks(task -> executor.submit((Callable<?>) task));
+    }
+  }
+
+  static void executeTwoTasks(ThrowingConsumer<JavaAsyncChild> task) {
+    testing.runWithSpan(
+        "parent",
+        () -> {
+          // this child will have a span
+          JavaAsyncChild child1 = new JavaAsyncChild();
+          // this child won't
+          JavaAsyncChild child2 = new JavaAsyncChild(false, false);
+          try {
+            task.accept(child1);
+            task.accept(child2);
+          } catch (Throwable t) {
+            throw new AssertionError(t);
+          }
+          child1.waitForCompletion();
+          child2.waitForCompletion();
+        });
+    testing.waitAndAssertTraces(
+        trace ->
+            trace.hasSpansSatisfyingExactly(
+                span -> span.hasName("parent").hasKind(SpanKind.INTERNAL).hasNoParent(),
+                span ->
+                    span.hasName("asyncChild")
+                        .hasKind(SpanKind.INTERNAL)
+                        .hasParent(trace.getSpan(0))));
+  }
+
+  static void executeAndCancelTasks(Function<JavaAsyncChild, Future<?>> task) {
+    List<JavaAsyncChild> children = new ArrayList<>();
+    List<Future<?>> jobFutures = new ArrayList<>();
+
+    testing.runWithSpan(
+        "parent",
+        () -> {
+          for (int i = 0; i < 20; i++) {
+            // Our current instrumentation instrumentation does not behave very well
+            // if we try to reuse Callable/Runnable. Namely we would be getting 'orphaned'
+            // child traces sometimes since state can contain only one parent span - and
+            // we do not really have a good way for attributing work to correct parent span
+            // if we reuse Callable/Runnable.
+            // Solution for now is to never reuse a Callable/Runnable.
+            JavaAsyncChild child = new JavaAsyncChild(true, true);
+            children.add(child);
+            Future<?> f = task.apply(child);
+            jobFutures.add(f);
+          }
+
+          jobFutures.forEach(f -> f.cancel(false));
+          children.forEach(JavaAsyncChild::unblock);
+        });
+
+    // Just check there is a single trace, this test is primarily to make sure that scopes aren't
+    // leak on
+    // cancellation.
+    testing.waitAndAssertTraces(trace -> {});
+  }
+
+  @SuppressWarnings("RedundantOverride")
+  private static class CustomThreadPoolExecutor extends AbstractExecutorService {
+
+    private volatile boolean running = true;
+
+    private final BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(20);
+
+    private final Thread workerThread =
+        new Thread(
+            () -> {
+              try {
+                while (running) {
+                  Runnable runnable = workQueue.take();
+                  runnable.run();
+                }
+              } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                throw new AssertionError(e);
+              } catch (Throwable t) {
+                throw new AssertionError(t);
+              }
+            },
+            "ExecutorTestThread");
+
+    private CustomThreadPoolExecutor() {
+      workerThread.start();
+    }
+
+    @Override
+    public void shutdown() {
+      running = false;
+      workerThread.interrupt();
+    }
+
+    @Override
+    public List<Runnable> shutdownNow() {
+      running = false;
+      workerThread.interrupt();
+      return Collections.emptyList();
+    }
+
+    @Override
+    public boolean isShutdown() {
+      return !running;
+    }
+
+    @Override
+    public boolean isTerminated() {
+      return workerThread.isAlive();
+    }
+
+    @Override
+    public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
+      workerThread.join(unit.toMillis(timeout));
+      return true;
+    }
+
+    @Override
+    public <T> Future<T> submit(Callable<T> task) {
+      RunnableFuture<T> future = newTaskFor(task);
+      execute(future);
+      return future;
+    }
+
+    @Override
+    public <T> Future<T> submit(Runnable task, T result) {
+      RunnableFuture<T> future = newTaskFor(task, result);
+      execute(future);
+      return future;
+    }
+
+    @Override
+    public Future<?> submit(Runnable task) {
+      RunnableFuture<?> future = newTaskFor(task, null);
+      execute(future);
+      return future;
+    }
+
+    @Override
+    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) {
+      return Collections.singletonList(submit(tasks.stream().findFirst().get()));
+    }
+
+    @Override
+    public <T> List<Future<T>> invokeAll(
+        Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) {
+      return Collections.singletonList(submit(tasks.stream().findFirst().get()));
+    }
+
+    @Override
+    public <T> T invokeAny(Collection<? extends Callable<T>> tasks) {
+      submit(tasks.stream().findFirst().get());
+      return null;
+    }
+
+    @Override
+    public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) {
+      submit(tasks.stream().findFirst().get());
+      return null;
+    }
+
+    @Override
+    public void execute(Runnable command) {
+      workQueue.add(command);
+    }
+  }
+}

instrumentation/executors/javaagent/src/test/java/JavaAsyncChild.java → instrumentation/executors/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/javaconcurrent/JavaAsyncChild.java

@@ -3,6 +3,8 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
+package io.opentelemetry.javaagent.instrumentation.javaconcurrent;
+
 import io.opentelemetry.api.GlobalOpenTelemetry;
 import io.opentelemetry.api.trace.Tracer;
 import java.util.concurrent.Callable;
@@ -55,8 +57,13 @@ public class JavaAsyncChild extends ForkJoinTask<Object> implements Runnable, Ca
     return null;
   }
 
-  public void waitForCompletion() throws InterruptedException {
-    latch.await();
+  public void waitForCompletion() {
+    try {
+      latch.await();
+    } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
+      throw new AssertionError(e);
+    }
   }
 
   private void runImpl() {

instrumentation/executors/javaagent/src/test/java/LambdaGen.java → instrumentation/executors/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/javaconcurrent/LambdaGen.java

@@ -3,6 +3,8 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
+package io.opentelemetry.javaagent.instrumentation.javaconcurrent;
+
 import java.util.concurrent.Callable;
 
 class LambdaGen {

instrumentation/executors/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/javaconcurrent/ModuleInjectionTest.java

@@ -0,0 +1,22 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.javaconcurrent;
+
+import static org.assertj.core.api.Assertions.assertThatCode;
+
+import javax.swing.RepaintManager;
+import org.junit.jupiter.api.Test;
+
+// This class tests that we correctly add module references when instrumenting
+class ModuleInjectionTest {
+
+  // There's nothing special about RepaintManager other than it's in a module (java.desktop) that
+  // doesn't read the "unnamed module" and it creates an instrumented runnable in its constructor.
+  @Test
+  void instrumentingJavaDesktopClass() {
+    assertThatCode(RepaintManager::new).doesNotThrowAnyException();
+  }
+}

instrumentation/executors/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/javaconcurrent/StreamTest.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.javaconcurrent;
+
+import io.opentelemetry.instrumentation.testing.junit.AgentInstrumentationExtension;
+import io.opentelemetry.instrumentation.testing.junit.InstrumentationExtension;
+import io.opentelemetry.sdk.testing.assertj.SpanDataAssert;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.Consumer;
+import java.util.stream.IntStream;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+class StreamTest {
+
+  @RegisterExtension
+  static final InstrumentationExtension testing = AgentInstrumentationExtension.create();
+
+  @Test
+  void parallelStream() {
+    testing.runWithSpan(
+        "parent",
+        () ->
+            IntStream.range(0, 20)
+                .parallel()
+                .forEach(unused -> testing.runWithSpan("child", () -> {})));
+
+    testing.waitAndAssertTraces(
+        trace -> {
+          List<Consumer<SpanDataAssert>> assertions = new ArrayList<>();
+          assertions.add(span -> span.hasName("parent").hasNoParent());
+          IntStream.range(0, 20)
+              .forEach(
+                  unused ->
+                      assertions.add(span -> span.hasName("child").hasParent(trace.getSpan(0))));
+
+          trace.hasSpansSatisfyingExactly(assertions);
+        });
+  }
+}