From 69ec01ac01127d3a1805a97afc928ebd0029c9d9 Mon Sep 17 00:00:00 2001 From: Oliver Drotbohm Date: Tue, 28 Nov 2023 15:27:17 +0100 Subject: [PATCH] GH-399 - Avoid event publication completion for failed CompletableFutures. We now explicitly handle CompletableFuture instances returned from transactional event listeners by registering the completion handlers only on success, the debug logging on failure and immediately return the decorated instance. Previously, a failed CompletableFuture instance would still have the publication marked completed as it doesn't cause any exception being thrown and thus ultimately ending up in the code path that issues the completion. --- .../support/CompletionRegisteringAdvisor.java | 50 ++++++++++++++---- ...CompletionRegisteringAdvisorUnitTests.java | 52 ++++++++++++++++++- 2 files changed, 89 insertions(+), 13 deletions(-) diff --git a/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/support/CompletionRegisteringAdvisor.java b/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/support/CompletionRegisteringAdvisor.java index 36054b8a..c1c75614 100644 --- a/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/support/CompletionRegisteringAdvisor.java +++ b/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/support/CompletionRegisteringAdvisor.java @@ -16,6 +16,7 @@ package org.springframework.modulith.events.support; import java.lang.reflect.Method; +import java.util.concurrent.CompletableFuture; import java.util.function.Supplier; import org.aopalliance.aop.Advice; @@ -31,6 +32,7 @@ import org.springframework.aop.support.annotation.AnnotationMatchingPointcut; import org.springframework.core.Ordered; import org.springframework.core.annotation.AnnotatedElementUtils; import org.springframework.lang.NonNull; +import org.springframework.lang.Nullable; import org.springframework.modulith.events.core.EventPublicationRegistry; import org.springframework.modulith.events.core.PublicationTargetIdentifier; import org.springframework.transaction.event.TransactionPhase; @@ -164,25 +166,30 @@ public class CompletionRegisteringAdvisor extends AbstractPointcutAdvisor { Object result = null; var method = invocation.getMethod(); + var argument = invocation.getArguments()[0]; try { - result = invocation.proceed(); - } catch (Exception o_O) { - if (LOG.isDebugEnabled()) { - LOG.debug("Invocation of listener {} failed. Leaving event publication uncompleted.", method, o_O); - } else { - LOG.info("Invocation of listener {} failed with message {}. Leaving event publication uncompleted.", - method, o_O.getMessage()); + result = invocation.proceed(); + + if (result instanceof CompletableFuture future) { + + return future + .thenAccept(it -> markCompleted(method, argument)) + .exceptionallyCompose(it -> { + handleFailure(method, it); + return CompletableFuture.failedFuture(it); + }); } + } catch (Throwable o_O) { + + handleFailure(method, o_O); + throw o_O; } - // Mark publication complete if the method is a transactional event listener. - String adapterId = ADAPTERS.get(method).getListenerId(); - PublicationTargetIdentifier identifier = PublicationTargetIdentifier.of(adapterId); - registry.get().markCompleted(invocation.getArguments()[0], identifier); + markCompleted(method, argument); return result; } @@ -196,6 +203,27 @@ public class CompletionRegisteringAdvisor extends AbstractPointcutAdvisor { return Ordered.HIGHEST_PRECEDENCE + 10; } + @Nullable + private static Void handleFailure(Method method, Throwable o_O) { + + if (LOG.isDebugEnabled()) { + LOG.debug("Invocation of listener {} failed. Leaving event publication uncompleted.", method, o_O); + } else { + LOG.info("Invocation of listener {} failed with message {}. Leaving event publication uncompleted.", + method, o_O.getMessage()); + } + + return null; + } + + private void markCompleted(Method method, Object event) { + + // Mark publication complete if the method is a transactional event listener. + String adapterId = ADAPTERS.get(method).getListenerId(); + PublicationTargetIdentifier identifier = PublicationTargetIdentifier.of(adapterId); + registry.get().markCompleted(event, identifier); + } + private static TransactionalApplicationListenerMethodAdapter createAdapter(Method method) { return new TransactionalApplicationListenerMethodAdapter(null, method.getDeclaringClass(), method); } diff --git a/spring-modulith-events/spring-modulith-events-core/src/test/java/org/springframework/modulith/events/support/CompletionRegisteringAdvisorUnitTests.java b/spring-modulith-events/spring-modulith-events-core/src/test/java/org/springframework/modulith/events/support/CompletionRegisteringAdvisorUnitTests.java index 9197638a..915b9ebc 100644 --- a/spring-modulith-events/spring-modulith-events-core/src/test/java/org/springframework/modulith/events/support/CompletionRegisteringAdvisorUnitTests.java +++ b/spring-modulith-events/spring-modulith-events-core/src/test/java/org/springframework/modulith/events/support/CompletionRegisteringAdvisorUnitTests.java @@ -19,6 +19,7 @@ import static org.assertj.core.api.Assertions.*; import static org.mockito.ArgumentMatchers.*; import static org.mockito.Mockito.*; +import java.util.concurrent.CompletableFuture; import java.util.function.BiConsumer; import org.junit.jupiter.api.Test; @@ -26,6 +27,7 @@ import org.springframework.aop.framework.Advised; import org.springframework.aop.framework.ProxyFactory; import org.springframework.context.event.EventListener; import org.springframework.modulith.events.core.EventPublicationRegistry; +import org.springframework.scheduling.annotation.Async; import org.springframework.transaction.event.TransactionPhase; import org.springframework.transaction.event.TransactionalEventListener; @@ -59,6 +61,34 @@ class CompletionRegisteringAdvisorUnitTests { assertNonCompletion(SomeEventListener::nonEventListener); } + @Test // GH-399 + void doesNotTriggerCompletionOnFailedCompletableFuture() throws Throwable { + + var result = createProxyFor(bean).asyncWithResult(true); + + assertThat(result.isDone()).isFalse(); + verify(registry, never()).markCompleted(any(), any()); + + Thread.sleep(500); + + assertThat(result.isCompletedExceptionally()).isTrue(); + verify(registry, never()).markCompleted(any(), any()); + } + + @Test // GH-399 + void marksLazilyComputedCompletableFutureAsCompleted() throws Throwable { + + var result = createProxyFor(bean).asyncWithResult(false); + + assertThat(result.isDone()).isFalse(); + verify(registry, never()).markCompleted(any(), any()); + + Thread.sleep(500); + + assertThat(result.isCompletedExceptionally()).isFalse(); + verify(registry).markCompleted(any(), any()); + } + private void assertCompletion(BiConsumer consumer) { assertCompletion(consumer, true); } @@ -78,11 +108,12 @@ class CompletionRegisteringAdvisorUnitTests { verify(registry, times(expected ? 1 : 0)).markCompleted(any(), any()); } - private Object createProxyFor(Object bean) { + @SuppressWarnings("unchecked") + private T createProxyFor(T bean) { ProxyFactory factory = new ProxyFactory(bean); factory.addAdvisor(new CompletionRegisteringAdvisor(() -> registry)); - return factory.getProxy(); + return (T) factory.getProxy(); } static class SomeEventListener { @@ -97,5 +128,22 @@ class CompletionRegisteringAdvisorUnitTests { void simpleEventListener(Object object) {} void nonEventListener(Object object) {} + + @Async + @TransactionalEventListener + CompletableFuture asyncWithResult(boolean fail) { + + return CompletableFuture.completedFuture(new Object()) + .thenComposeAsync(it -> { + + try { + Thread.sleep(200); + } catch (InterruptedException e) {} + + return fail + ? CompletableFuture.failedFuture(new IllegalArgumentException()) + : CompletableFuture.completedFuture(it); + }); + } } }