Add scala Future test case using multiple threads

apm-agent-plugins/apm-scala-concurrent-plugin/pom.xml

@@ -20,12 +20,17 @@
             <groupId>${project.groupId}</groupId>
             <artifactId>apm-java-concurrent-plugin</artifactId>
             <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.scala-lang</groupId>
+            <artifactId>scala-library</artifactId>
+            <version>2.13.6</version>
             <scope>test</scope>
         </dependency>
         <dependency>
             <groupId>org.scalameta</groupId>
             <artifactId>munit_2.13</artifactId>
-            <version>0.7.9</version>
+            <version>0.7.29</version>
             <scope>test</scope>
         </dependency>
     </dependencies>
@@ -36,9 +41,9 @@
             <plugin>
                 <groupId>net.alchim31.maven</groupId>
                 <artifactId>scala-maven-plugin</artifactId>
-                <version>4.3.1</version>
+                <version>4.5.4</version>
                 <configuration>
-                    <scalaVersion>2.13.2</scalaVersion>
+                    <scalaVersion>2.13.6</scalaVersion>
                 </configuration>
                 <executions>
                     <execution>

apm-agent-plugins/apm-scala-concurrent-plugin/src/main/java/co/elastic/apm/agent/scalaconcurrent/FutureInstrumentation.java

@@ -19,7 +19,9 @@
 package co.elastic.apm.agent.scalaconcurrent;
 
 import co.elastic.apm.agent.bci.TracerAwareInstrumentation;
+import co.elastic.apm.agent.concurrent.JavaConcurrent;
 import co.elastic.apm.agent.impl.transaction.AbstractSpan;
+import co.elastic.apm.agent.sdk.advice.AssignTo;
 import co.elastic.apm.agent.sdk.weakconcurrent.WeakConcurrent;
 import co.elastic.apm.agent.sdk.weakconcurrent.WeakMap;
 import net.bytebuddy.asm.Advice;
@@ -32,9 +34,11 @@
 import java.util.Arrays;
 import java.util.Collection;
 
+import static net.bytebuddy.matcher.ElementMatchers.hasSuperType;
 import static net.bytebuddy.matcher.ElementMatchers.isConstructor;
 import static net.bytebuddy.matcher.ElementMatchers.named;
 import static net.bytebuddy.matcher.ElementMatchers.returns;
+import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
 
 public abstract class FutureInstrumentation extends TracerAwareInstrumentation {
 
@@ -48,7 +52,34 @@ public Collection<String> getInstrumentationGroupNames() {
         return Arrays.asList("scala-future", "experimental");
     }
 
-    public static class ConstructorInstrumentation extends FutureInstrumentation {
+    public static class BatchedExecutionContextInstrumentation extends FutureInstrumentation {
+
+        @Override
+        public ElementMatcher<? super TypeDescription> getTypeMatcher() {
+            return hasSuperType(named("scala.concurrent.BatchingExecutor"));
+        }
+
+        @Override
+        public ElementMatcher<? super MethodDescription> getMethodMatcher() {
+            return named("submitForExecution").and(returns(void.class)).and(takesArguments(Runnable.class));
+        }
+
+        @Nullable
+        @AssignTo.Argument(0)
+        @Advice.OnMethodEnter(suppress = Throwable.class, inline = false)
+        public static Runnable onExecute(@Advice.Argument(0) @Nullable Runnable runnable) {
+            return JavaConcurrent.withContext(runnable, tracer);
+        }
+
+        @Advice.OnMethodExit(suppress = Throwable.class, onThrowable = Throwable.class, inline = false)
+        public static void onExit(@Nullable @Advice.Thrown Throwable thrown,
+                                  @Advice.Argument(value = 0) @Nullable Runnable runnable) {
+            JavaConcurrent.doFinally(thrown, runnable);
+        }
+    }
+
+
+    public static class TransformationConstructorInstrumentation extends FutureInstrumentation {
 
         @Override
         public ElementMatcher<? super TypeDescription> getTypeMatcher() {
@@ -69,13 +100,15 @@ public static void onExit(@Advice.This Object thiz) {
                     // this span might be ended before the Promise$Transformation#run method starts
                     // we have to avoid that this span gets recycled, even in the above mentioned case
                     context.incrementReferences();
+                    // Do no discard branches leading to async operations so not to break span references
+                    context.setNonDiscardable();
                 }
             }
         }
 
     }
 
-    public static class RunInstrumentation extends FutureInstrumentation {
+    public static class TransformationRunInstrumentation extends FutureInstrumentation {
 
         @Override
         public ElementMatcher<? super TypeDescription> getTypeMatcher() {

apm-agent-plugins/apm-scala-concurrent-plugin/src/main/resources/META-INF/services/co.elastic.apm.agent.sdk.ElasticApmInstrumentation

@@ -1,2 +1,3 @@
-co.elastic.apm.agent.scalaconcurrent.FutureInstrumentation$ConstructorInstrumentation
-co.elastic.apm.agent.scalaconcurrent.FutureInstrumentation$RunInstrumentation
+co.elastic.apm.agent.scalaconcurrent.FutureInstrumentation$BatchedExecutionContextInstrumentation
+co.elastic.apm.agent.scalaconcurrent.FutureInstrumentation$TransformationConstructorInstrumentation
+co.elastic.apm.agent.scalaconcurrent.FutureInstrumentation$TransformationRunInstrumentation