Excluding Spring-JMS specific Runnable from the concurrent plugin (#1496)

Author: eyalkoren

CHANGELOG.asciidoc

@@ -54,6 +54,7 @@ endif::[]
 * Fix `IllegalAccessError: Module 'java.base' no access to: package 'java.lang'...` in J9 VMs of Java version >= 9 -
   {pull}1468[#1468]
 * Fix JVM version parsing on HP-UX {pull}1477[#1477]
+* Fix Spring-JMS transactions lifecycle management when using multiple concurrent consumers - {pull}1496[#1496]
 
 ===== Refactors
 * Migrate some plugins to indy dispatcher {pull}1404[1404] {pull}1411[1411]

apm-agent-core/src/main/java/co/elastic/apm/agent/impl/transaction/AbstractSpan.java

@@ -194,6 +194,10 @@ public boolean isReferenced() {
         return references.get() > 0;
     }
 
+    public boolean isFinished() {
+        return finished;
+    }
+
     /**
      * How long the transaction took to complete, in µs
      */

apm-agent-plugins/apm-java-concurrent-plugin/src/main/java/co/elastic/apm/agent/concurrent/JavaConcurrent.java

@@ -35,7 +35,9 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Set;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ForkJoinTask;
 
@@ -46,13 +48,29 @@ public class JavaConcurrent {
         <Class<? extends ElasticApmInstrumentation>>singletonList(RunnableCallableForkJoinTaskInstrumentation.class);
     static final ThreadLocal<Boolean> needsContext = new ThreadLocal<>();
 
+    private static final Set<String> EXCLUDED_EXECUTABLE_TYPES;
+
+    static {
+        EXCLUDED_EXECUTABLE_TYPES = new HashSet<String>();
+        EXCLUDED_EXECUTABLE_TYPES.add(RunnableLambdaWrapper.class.getName());
+        EXCLUDED_EXECUTABLE_TYPES.add(CallableLambdaWrapper.class.getName());
+        // Spring-JMS polling mechanism that translates to passive onMessage handling
+        EXCLUDED_EXECUTABLE_TYPES.add("org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker");
+    }
+
     private static void removeContext(Object o) {
         AbstractSpan<?> context = contextMap.remove(o);
         if (context != null) {
             context.decrementReferences();
         }
     }
 
+    private static boolean shouldAvoidContextPropagation(@Nullable Object executable) {
+        return executable == null ||
+            EXCLUDED_EXECUTABLE_TYPES.contains(executable.getClass().getName()) ||
+            needsContext.get() == Boolean.FALSE;
+    }
+
     @Nullable
     public static AbstractSpan<?> restoreContext(Object o, Tracer tracer) {
         // When an Executor executes directly on the current thread we need to enable this thread for context propagation again
@@ -76,7 +94,7 @@ public static AbstractSpan<?> restoreContext(Object o, Tracer tracer) {
      */
     @Nullable
     public static Runnable withContext(@Nullable Runnable runnable, Tracer tracer) {
-        if (runnable instanceof RunnableLambdaWrapper || runnable == null || needsContext.get() == Boolean.FALSE) {
+        if (shouldAvoidContextPropagation(runnable)) {
             return runnable;
         }
         needsContext.set(Boolean.FALSE);
@@ -104,7 +122,7 @@ private static void captureContext(Object task, AbstractSpan<?> active) {
      */
     @Nullable
     public static <T> Callable<T> withContext(@Nullable Callable<T> callable, Tracer tracer) {
-        if (callable instanceof CallableLambdaWrapper || callable == null || needsContext.get() == Boolean.FALSE) {
+        if (shouldAvoidContextPropagation(callable)) {
             return callable;
         }
         needsContext.set(Boolean.FALSE);
@@ -121,7 +139,7 @@ public static <T> Callable<T> withContext(@Nullable Callable<T> callable, Tracer
 
     @Nullable
     public static <T> ForkJoinTask<T> withContext(@Nullable ForkJoinTask<T> task, Tracer tracer) {
-        if (task == null || needsContext.get() == Boolean.FALSE) {
+        if (shouldAvoidContextPropagation(task)) {
             return task;
         }
         needsContext.set(Boolean.FALSE);

apm-agent-plugins/apm-jms-plugin/apm-jms-plugin-base/pom.xml

@@ -0,0 +1,19 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <artifactId>apm-jms-plugin</artifactId>
+        <groupId>co.elastic.apm</groupId>
+        <version>1.18.2-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>apm-jms-plugin-base</artifactId>
+    <name>${project.groupId}:${project.artifactId}</name>
+
+    <properties>
+        <apm-agent-parent.base.dir>${project.basedir}/../../..</apm-agent-parent.base.dir>
+    </properties>
+
+</project>

apm-agent-plugins/apm-jms-plugin/src/main/java/co/elastic/apm/agent/jms/BaseJmsInstrumentation.java renamed to apm-agent-plugins/apm-jms-plugin/apm-jms-plugin-base/src/main/java/co/elastic/apm/agent/jms/BaseJmsInstrumentation.java

@@ -61,11 +61,7 @@ public JmsMessageListenerInstrumentation(ElasticApmTracer tracer) {
 
     @Override
     public ElementMatcher<? super TypeDescription> getTypeMatcher() {
-        return not(isInterface())
-            .and(
-                hasSuperType(named("javax.jms.MessageListener"))
-                    .or(hasSuperType(named("org.springframework.jms.listener.SessionAwareMessageListener")))
-            );
+        return not(isInterface()).and(hasSuperType(named("javax.jms.MessageListener")));
     }
 
     @Override