Fix: Spring 7 Webflux NoSuchElementError on HttpHeaders#entrySet()

CHANGELOG.next-release.md

@@ -11,6 +11,7 @@ This file contains all changes which are not released yet.
 <!--FIXES-START-->
 
 - avoid caching when reading version from jar to prevent side effects - [#4543](https://github.com/elastic/apm-agent-java/pull/4543)
+- fix Spring Webflux 7 NoSuchMethodError on HttpHeaders#entrySet() - [#4556](https://github.com/elastic/apm-agent-java/pull/4556)
 
 <!--FIXES-END-->
 # Features and enhancements

apm-agent-core/src/test/java/co/elastic/apm/agent/testutils/Java17OnlyTest.java

@@ -35,21 +35,21 @@
  * Why not just put @EnabledForJreRange directly on the test classes?
  * JUnit reflectively loads the target class when discovering tests.
  * Because of spring, the tests contain references/annotations compiled with Java 17.
- * This in turn leads to UnsupportedClassVersionErrors before JUnit can evaluate the @EnableForJRERange when running on older java versions (e.g. 11).
+ * This in turn leads to UnsupportedClassVersionErrors before JUnit can evaluate the @EnabledForJreRange when running on older java versions (e.g. 11).
  * <p>
  * Therefore, this class can be used to wrap tests, as it programmatically triggers the test execution.
  * The actual test implementation should not be named *Test to not be discovered by the maven surefire plugin.
  */
 public abstract class Java17OnlyTest {
 
-    private Class<?> actualTestClass;
+    private final Class<?> actualTestClass;
 
     public Java17OnlyTest(Class<?> testClazz) {
         this.actualTestClass = testClazz;
     }
 
-    @EnabledForJreRange(min = JRE.JAVA_17)
     @Test
+    @EnabledForJreRange(min = JRE.JAVA_17, max = JRE.JAVA_25)
     public void runTests() {
         LauncherDiscoveryRequest request = LauncherDiscoveryRequestBuilder.request()
             .selectors(selectClass(actualTestClass))

apm-agent-plugins/apm-reactor-plugin/src/main/java/co/elastic/apm/agent/reactor/SubscriptionCancelInstrumentation.java

@@ -0,0 +1,82 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package co.elastic.apm.agent.reactor;
+
+import co.elastic.apm.agent.sdk.ElasticApmInstrumentation;
+import net.bytebuddy.asm.Advice;
+import net.bytebuddy.description.NamedElement;
+import net.bytebuddy.description.method.MethodDescription;
+import net.bytebuddy.description.type.TypeDescription;
+import net.bytebuddy.matcher.ElementMatcher;
+import org.reactivestreams.Subscription;
+
+import java.util.Collection;
+import java.util.Collections;
+
+import static co.elastic.apm.agent.sdk.bytebuddy.CustomElementMatchers.classLoaderCanLoadClass;
+import static net.bytebuddy.matcher.ElementMatchers.declaresMethod;
+import static net.bytebuddy.matcher.ElementMatchers.hasSuperType;
+import static net.bytebuddy.matcher.ElementMatchers.isInterface;
+import static net.bytebuddy.matcher.ElementMatchers.nameContains;
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static net.bytebuddy.matcher.ElementMatchers.not;
+import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
+
+public class SubscriptionCancelInstrumentation extends ElasticApmInstrumentation {
+
+    @Override
+    public ElementMatcher.Junction<ClassLoader> getClassLoaderMatcher() {
+        return classLoaderCanLoadClass("reactor.core.CoreSubscriber");
+    }
+
+    @Override
+    public ElementMatcher<? super NamedElement> getTypeMatcherPreFilter() {
+        return nameContains("Subscriber").or(nameContains("Subscription"));
+    }
+
+    @Override
+    public ElementMatcher<? super TypeDescription> getTypeMatcher() {
+        return not(isInterface())
+            .and(declaresMethod(getMethodMatcher()))
+            .and(hasSuperType(named("org.reactivestreams.Subscription")));
+    }
+
+    @Override
+    public ElementMatcher<? super MethodDescription> getMethodMatcher() {
+        return named("cancel").and(takesArguments(0));
+    }
+
+    @Override
+    public Collection<String> getInstrumentationGroupNames() {
+        return Collections.singleton("reactor");
+    }
+
+    @Override
+    public String getAdviceClassName() {
+        return "co.elastic.apm.agent.reactor.SubscriptionCancelInstrumentation$CancelAdvice";
+    }
+
+    public static class CancelAdvice {
+
+        @Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class, inline = false)
+        public static void afterCancel(@Advice.This Subscription subscription) {
+            TracedSubscriber.onCancel(subscription);
+        }
+    }
+}

apm-agent-plugins/apm-reactor-plugin/src/main/java/co/elastic/apm/agent/reactor/TracedSubscriber.java

@@ -23,6 +23,7 @@
 import co.elastic.apm.agent.sdk.state.GlobalVariables;
 import co.elastic.apm.agent.sdk.weakconcurrent.WeakConcurrent;
 import co.elastic.apm.agent.sdk.weakconcurrent.WeakMap;
+import co.elastic.apm.agent.sdk.weakconcurrent.WeakSet;
 import co.elastic.apm.agent.tracer.TraceState;
 import co.elastic.apm.agent.tracer.GlobalTracer;
 import co.elastic.apm.agent.tracer.Tracer;
@@ -49,6 +50,8 @@ public class TracedSubscriber<T> implements CoreSubscriber<T> {
 
     private static final ReferenceCountedMap<TracedSubscriber<?>, TraceState<?>> contextMap = GlobalTracer.get().newReferenceCountedMap();
 
+    private static final WeakMap<Subscription, WeakSet<TracedSubscriber<?>>> subscriptionMap = WeakConcurrent.buildMap();
+
     private static final String HOOK_KEY = "elastic-apm";
 
     private final CoreSubscriber<? super T> subscriber;
@@ -82,6 +85,7 @@ public void onSubscribe(Subscription s) {
         boolean hasActivated = doEnter("onSubscribe", context);
         Throwable thrown = null;
         try {
+            registerSubscription(s);
             subscriber.onSubscribe(s);
         } catch (Throwable e) {
             thrown = e;
@@ -187,6 +191,32 @@ private void doExit(boolean deactivate, String method, @Nullable TraceState<?> c
         context.deactivate();
     }
 
+    private void registerSubscription(Subscription subscription) {
+        WeakSet<TracedSubscriber<?>> subscribers = subscriptionMap.get(subscription);
+        if (subscribers == null) {
+            WeakSet<TracedSubscriber<?>> newSubscribers = WeakConcurrent.buildSet();
+            subscribers = subscriptionMap.putIfAbsent(subscription, newSubscribers);
+            if (subscribers == null) {
+                subscribers = newSubscribers;
+            }
+        }
+        subscribers.add(this);
+    }
+
+    /**
+     * Cancellation does not emit a terminal signal, so it is observed by
+     * {@link SubscriptionCancelInstrumentation} instead.
+     */
+    static void onCancel(Subscription subscription) {
+        WeakSet<TracedSubscriber<?>> subscribers = subscriptionMap.remove(subscription);
+        if (subscribers == null) {
+            return;
+        }
+        for (TracedSubscriber<?> subscriber : subscribers) {
+            subscriber.discardIf(true);
+        }
+    }
+
     private void discardIf(boolean condition) {
         if (!condition) {
             return;

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

@@ -1 +1,2 @@
 co.elastic.apm.agent.reactor.ReactorInstrumentation
+co.elastic.apm.agent.reactor.SubscriptionCancelInstrumentation

apm-agent-plugins/apm-reactor-plugin/src/test/java/co/elastic/apm/agent/reactor/TracedSubscriberTest.java

@@ -91,7 +91,10 @@ void after() {
 
         // ensure clean as new hooks setup for all tests (some might have removed it)
         TracedSubscriber.unregisterHooks();
-        TracedSubscriber.registerHooks(tracer);
+        // Register through ReactorInstrumentation so the hook and the cancel advice use
+        // TracedSubscriber classes loaded by the same plugin class loader.
+        Mono.just(1);
+        checkHookRegistration(true, "hook should be registered automatically after each test");
 
         flushGcExpiry(3);
     }
@@ -223,6 +226,21 @@ void contextPropagation_Flux_error() {
             .verifyErrorMatches(t -> t == error);
     }
 
+    @Test
+    void cancelledSubscription_releasedContext() {
+        transaction = startTestRootTransaction("root");
+        int initialReferenceCount = transaction.getReferenceCount();
+
+        Flux<Integer> flux = Flux.just(1, 2, 3)
+            .subscribeOn(SUBSCRIBE_SCHEDULER);
+
+        StepVerifier.create(flux.log())
+            .thenCancel()
+            .verify();
+
+        assertThat(transaction.getReferenceCount()).isEqualTo(initialReferenceCount);
+    }
+
     @Test
     void ignoreNoActiveContext() {
         assertThat(tracer.getActive()).isNull();

apm-agent-plugins/apm-spring-webflux/apm-spring-webflux-plugin-spring7/pom.xml

@@ -0,0 +1,112 @@
+<?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>
+        <groupId>co.elastic.apm</groupId>
+        <artifactId>apm-spring-webflux</artifactId>
+        <version>1.56.1-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>apm-spring-webflux-plugin-spring7</artifactId>
+    <name>${project.groupId}:${project.artifactId}</name>
+
+    <properties>
+        <!-- for licence header plugin -->
+        <apm-agent-parent.base.dir>${project.basedir}/../../..</apm-agent-parent.base.dir>
+
+        <animal.sniffer.skip>true</animal.sniffer.skip>
+    </properties>
+
+    <dependencyManagement>
+        <dependencies>
+            <dependency>
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-dependencies</artifactId>
+                <version>${version.spring-boot-4}</version>
+                <type>pom</type>
+                <scope>import</scope>
+            </dependency>
+        </dependencies>
+    </dependencyManagement>
+
+    <dependencies>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>apm-spring-webflux-spring5</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <dependency>
+            <!-- through the spring-boot dependency management, the test app will be "updated" to spring boot 4 -->
+            <groupId>${project.groupId}</groupId>
+            <artifactId>apm-spring-webflux-testapp-spring7</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework</groupId>
+            <artifactId>spring-web</artifactId>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.projectreactor</groupId>
+            <artifactId>reactor-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>co.elastic.apm</groupId>
+            <artifactId>apm-spring-webflux-spring5</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+            <type>test-jar</type>
+        </dependency>
+
+        <!-- required for context-propagation during tests, but only at runtime -->
+        <dependency>
+            <groupId>co.elastic.apm</groupId>
+            <artifactId>apm-reactor-plugin</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>co.elastic.apm</groupId>
+            <artifactId>apm-reactor-plugin</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+            <type>test-jar</type>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.ivy</groupId>
+            <artifactId>ivy</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-dependency-plugin</artifactId>
+            </plugin>
+        </plugins>
+    </build>
+
+    <profiles>
+        <profile>
+            <!-- Spring Boot 4 / Spring Framework 7 requires junit 6 which requires java 17 -->
+            <id>testing-jdk-11</id>
+            <activation>
+                <property>
+                    <name>test_java_version</name>
+                    <value>11</value>
+                </property>
+            </activation>
+            <properties>
+                <skipTests>true</skipTests>
+            </properties>
+        </profile>
+    </profiles>
+
+</project>

apm-agent-plugins/apm-spring-webflux/apm-spring-webflux-plugin-spring7/src/test/java/co/elastic/apm/agent/springwebflux/Spring7HeaderGetterTest.java

@@ -0,0 +1,32 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package co.elastic.apm.agent.springwebflux;
+
+import co.elastic.apm.agent.testutils.Java17OnlyTest;
+
+public class Spring7HeaderGetterTest extends Java17OnlyTest {
+
+    public Spring7HeaderGetterTest() {
+        super(Impl.class);
+    }
+
+    public static class Impl extends HeaderGetterTest {
+
+    }
+}

apm-agent-plugins/apm-spring-webflux/apm-spring-webflux-plugin-spring7/src/test/java/co/elastic/apm/agent/springwebflux/Spring7ServerAnnotatedInstrumentationTest.java

@@ -0,0 +1,32 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package co.elastic.apm.agent.springwebflux;
+
+
+import co.elastic.apm.agent.testutils.Java17OnlyTest;
+
+public class Spring7ServerAnnotatedInstrumentationTest extends Java17OnlyTest {
+
+    public Spring7ServerAnnotatedInstrumentationTest() {
+        super(Impl.class);
+    }
+
+    public static class Impl extends ServerAnnotatedInstrumentationTest {
+    }
+}