Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.next-release.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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))
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,72 @@
/*
* 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.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.hasSuperType;
import static net.bytebuddy.matcher.ElementMatchers.isInterface;
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 TypeDescription> getTypeMatcher() {

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is there a common name to all the implementations we need to instrument here ? the hasSuperType is very expensive without getTypeMatcherPreFilter, so if possible adding getTypeMatcherPreFilter would be relevant here.

@mtomik mtomik Aug 26, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

only I can think of is something like this, but that probably will not narrow it much, right?

    @Override
    public ElementMatcher<? super NamedElement> getTypeMatcherPreFilter() {
        return nameContains("Subscription").or(nameContains("Subscriber"));
    }

or maybe have multiple these instrumentations by package/library that implements the CoreSubscriber ? or have just one for the implementations from reactor.core and the rest will be cleaned potentially by the GC later?

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this should be fine as a first step, do you have a list of all the classes that are being currently instrumented here to verify all of them match ? If those are spring classes I would expect them to stay consistently named. If those are user-provided then it might be too narrow and we need another approach.

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

the problem is, that there can come any implementation of CoreSubscriber. if we miss some thanks to this filter, the TracedSubscriber will not remove it immediately from its contextMap / subscriptionMap.

if I understand it correctly, the memory leak caused by this was always handled by GC. in this newer spring 7 came newer reactor or other library that allowed us to spot it more easily? that was at least my thought...

I had here another solution to solve it - the wrapper around that subscription (ecde26e) , but I had there compile issue with the older java versions. so maybe just try to solve that to avoid using this instrumentation?

return not(isInterface()).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);
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand Down
Original file line number Diff line number Diff line change
@@ -1 +1,2 @@
co.elastic.apm.agent.reactor.ReactorInstrumentation
co.elastic.apm.agent.reactor.SubscriptionCancelInstrumentation
Original file line number Diff line number Diff line change
Expand Up @@ -223,6 +223,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()
Comment thread
jackshirazi marked this conversation as resolved.
.verify();

assertThat(transaction.getReferenceCount()).isEqualTo(initialReferenceCount);
}

@Test
void ignoreNoActiveContext() {
assertThat(tracer.getActive()).isNull();
Expand Down
Original file line number Diff line number Diff line change
@@ -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>

Comment thread
jackshirazi marked this conversation as resolved.
<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>
Original file line number Diff line number Diff line change
@@ -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 {

}
}
Original file line number Diff line number Diff line change
@@ -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 {
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,31 @@
/*
* 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 Spring7ServerFunctionalInstrumentationTest extends Java17OnlyTest {

public Spring7ServerFunctionalInstrumentationTest() {
super(Impl.class);
}

public static class Impl extends ServerFunctionalInstrumentationTest {
}
}
Loading
Loading