Fix scope leak in ObservedAspect.

Removed 'scope.close' call from the future completion thread; added new test case ObservedAspectTests#parentObservationScopeShouldNotLeakToFutureCompletionThread.

Signed-off-by: Maksim Petelin <[email protected]>

Author: pema4

micrometer-observation/src/main/java/io/micrometer/observation/aop/ObservedAspect.java

@@ -147,16 +147,16 @@ public ObservedAspect(ObservationRegistry registry,
             try {
                 Object result = pjp.proceed();
                 if (result == null) {
-                    stopObservation(observation, scope, null);
+                    stopObservation(observation, null);
                     return result;
                 }
                 else {
                     CompletionStage<?> stage = (CompletionStage<?>) result;
-                    return stage.whenComplete((res, error) -> stopObservation(observation, scope, error));
+                    return stage.whenComplete((res, error) -> stopObservation(observation, error));
                 }
             }
             catch (Throwable error) {
-                stopObservation(observation, scope, error);
+                stopObservation(observation, error);
                 throw error;
             }
             finally {
@@ -187,11 +187,10 @@ private Method getMethod(ProceedingJoinPoint pjp) throws NoSuchMethodException {
         return method;
     }
 
-    private void stopObservation(Observation observation, Observation.Scope scope, @Nullable Throwable error) {
+    private void stopObservation(Observation observation, @Nullable Throwable error) {
         if (error != null) {
             observation.error(error);
         }
-        scope.close();
         observation.stop();
     }
 

samples/micrometer-samples-spring-framework6/src/test/java/io/micrometer/samples/spring6/aop/ObservedAspectTests.java

@@ -169,6 +169,26 @@ void annotatedAsyncCallShouldBeObservedAndErrorRecorded() {
             .isEqualTo(simulatedException);
     }
 
+    @Test
+    void observationShouldNotLeakToFutureCompletionThread() {
+        registry.observationConfig().observationHandler(new ObservationTextPublisher());
+
+        AspectJProxyFactory pf = new AspectJProxyFactory(new ObservedService());
+        pf.addAspect(new ObservedAspect(registry));
+
+        ObservedService service = pf.getProxy();
+        FakeAsyncTask fakeAsyncTask = new FakeAsyncTask("test-result");
+        Executor sameThreadExecutor = new ThreadPoolExecutor(1, 1, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>());
+        CompletableFuture<String> asyncResult = service
+            .parent(() -> service.asyncChild(fakeAsyncTask, sameThreadExecutor));
+        CompletableFuture<Void> asyncAssertion = asyncResult
+            .thenRunAsync(() -> assertThat(registry).doesNotHaveAnyRemainingCurrentObservation(), sameThreadExecutor);
+        fakeAsyncTask.proceed();
+        fakeAsyncTask.get();
+
+        assertThat(asyncAssertion).succeedsWithin(Duration.ofMillis(200));
+    }
+
     @Test
     void customObservationConventionShouldBeUsed() {
         registry.observationConfig().observationHandler(new ObservationTextPublisher());
@@ -401,6 +421,18 @@ CompletableFuture<String> async(FakeAsyncTask fakeAsyncTask) {
                     contextSnapshot.wrapExecutor(Executors.newSingleThreadExecutor()));
         }
 
+        @Observed(name = "test.async-child")
+        CompletableFuture<String> asyncChild(FakeAsyncTask fakeAsyncTask, Executor singleThreadExecutor) {
+            System.out.println("async-child");
+            return CompletableFuture.supplyAsync(fakeAsyncTask, singleThreadExecutor);
+        }
+
+        @Observed(name = "test.parent")
+        <T> T parent(Supplier<T> supplier) {
+            System.out.println("parent");
+            return supplier.get();
+        }
+
     }
 
     interface TestBeanInterface {