From 29bc84cd0b27c4e736dc7441d2d6f6d81768dce9 Mon Sep 17 00:00:00 2001 From: Oliver Drotbohm Date: Mon, 21 Oct 2024 13:49:29 +0200 Subject: [PATCH] GH-891 - DefaultEventPublicationRegistry now properly unregisters in-progress publication on failed resubmission. MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit DefaultEventPublicationRegistry.processIncompletePublications(…) now actively unregisters the publication from being considered in progress after completion (either successful or failed). While CompletionRegisteringAdvisor should take care of that on the target listeners we now leave the publications in progress in consistent state independent of the actual target being invoked. Decrease log level of the failed listener invocation as it's not unusual for the listener to fail. Improved PublicationsInProgress by switching to a concurrent map internally to avoid ConcurrentModificationExceptions in case of multiple threads. --- .../core/DefaultEventPublicationRegistry.java | 58 +++++++++++++++++-- ...aultEventPublicationRegistryUnitTests.java | 41 ++++++++++++- 2 files changed, 92 insertions(+), 7 deletions(-) diff --git a/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/DefaultEventPublicationRegistry.java b/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/DefaultEventPublicationRegistry.java index a4004a0f..9fc034f5 100644 --- a/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/DefaultEventPublicationRegistry.java +++ b/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/DefaultEventPublicationRegistry.java @@ -20,8 +20,10 @@ import java.time.Duration; import java.time.Instant; import java.util.Collection; import java.util.HashSet; +import java.util.Iterator; import java.util.Optional; import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; import java.util.function.Consumer; import java.util.function.Predicate; import java.util.stream.Stream; @@ -216,9 +218,12 @@ public class DefaultEventPublicationRegistry } catch (Exception o_O) { - if (LOGGER.isErrorEnabled()) { - LOGGER.error("Error republishing event publication " + it, o_O); + if (LOGGER.isInfoEnabled()) { + LOGGER.info("Error republishing event publication %s.".formatted(it), o_O); } + + } finally { + inProgress.unregister(it); } }); } @@ -249,6 +254,30 @@ public class DefaultEventPublicationRegistry } } + /** + * Returns all {@link PublicationsInProgress}. + * + * @return will never be {@literal null}. + * @since 1.3 + */ + PublicationsInProgress getPublicationsInProgress() { + return inProgress; + } + + /** + * Marks the given {@link TargetEventPublication} as failed. + * + * @param publication must not be {@literal null}. + * @see #markFailed(Object, PublicationTargetIdentifier) + * @since 1.3 + */ + void markFailed(TargetEventPublication publication) { + + Assert.notNull(publication, "TargetEventPublication must not be null!"); + + markFailed(publication.getEvent(), publication.getTargetIdentifier()); + } + private static String getConfirmationMessage(Collection publications) { var size = publications.size(); @@ -266,9 +295,9 @@ public class DefaultEventPublicationRegistry * @author Oliver Drotbohm * @since 1.3 */ - static class PublicationsInProgress { + static class PublicationsInProgress implements Iterable { - private final Set publications = new HashSet<>(); + private final Set publications = ConcurrentHashMap.newKeySet(); /** * Registers the given {@link TargetEventPublication} as currently processed. @@ -301,6 +330,18 @@ public class DefaultEventPublicationRegistry .ifPresent(publications::remove); } + /** + * Unregisters the {@link TargetEventPublication}.. + * + * @param publication must not be {@literal null}. + */ + void unregister(TargetEventPublication publication) { + + Assert.notNull(publication, "TargetEventPublication must not be null!"); + + publications.remove(publication); + } + /** * Returns the {@link TargetEventPublication} associated with the given event and * {@link PublicationTargetIdentifier}. @@ -318,5 +359,14 @@ public class DefaultEventPublicationRegistry .filter(it -> it.isAssociatedWith(event, identifier)) .findFirst(); } + + /* + * (non-Javadoc) + * @see java.lang.Iterable#iterator() + */ + @Override + public Iterator iterator() { + return new HashSet<>(publications).iterator(); + } } } diff --git a/spring-modulith-events/spring-modulith-events-core/src/test/java/org/springframework/modulith/events/core/DefaultEventPublicationRegistryUnitTests.java b/spring-modulith-events/spring-modulith-events-core/src/test/java/org/springframework/modulith/events/core/DefaultEventPublicationRegistryUnitTests.java index 6e67c749..b5cd77d6 100644 --- a/spring-modulith-events/spring-modulith-events-core/src/test/java/org/springframework/modulith/events/core/DefaultEventPublicationRegistryUnitTests.java +++ b/spring-modulith-events/spring-modulith-events-core/src/test/java/org/springframework/modulith/events/core/DefaultEventPublicationRegistryUnitTests.java @@ -23,6 +23,7 @@ import static org.mockito.Mockito.*; import java.time.Clock; import java.time.Instant; import java.time.ZoneId; +import java.util.function.Consumer; import java.util.stream.Stream; import org.junit.jupiter.api.Test; @@ -39,7 +40,6 @@ import org.mockito.junit.jupiter.MockitoExtension; class DefaultEventPublicationRegistryUnitTests { @Mock EventPublicationRepository repository; - @Mock Clock clock; @Test // GH-206 void usesCustomClockIfConfigured() { @@ -47,9 +47,8 @@ class DefaultEventPublicationRegistryUnitTests { when(repository.create(any())).then(returnsFirstArg()); var now = Instant.now(); - var clock = Clock.fixed(now, ZoneId.systemDefault()); - var registry = new DefaultEventPublicationRegistry(repository, clock); + var registry = createRegistry(now); var identifier = PublicationTargetIdentifier.of("id"); var publications = registry.store(new Object(), Stream.of(identifier)); @@ -59,4 +58,40 @@ class DefaultEventPublicationRegistryUnitTests { assertThat(it.getTargetIdentifier()).isEqualTo(identifier); }); } + + @Test // GH-819 + void removesFailingResubmissionFromInProgressPublications() { + + when(repository.create(any())).then(returnsFirstArg()); + + var registry = createRegistry(Instant.now()); + var identifier = PublicationTargetIdentifier.of("id"); + + var failedPublications = registry.store(new Object(), Stream.of(identifier)).stream() + .peek(registry::markFailed) + .toList(); + + // Failed completions are not present in the in progress ones + assertThat(registry.getPublicationsInProgress()).isEmpty(); + + when(repository.findIncompletePublications()).thenReturn(failedPublications); + + registry.processIncompletePublications(__ -> true, failingConsumer(), null); + + // Failed re-submissions are not held in the in progress ones, either. + assertThat(registry.getPublicationsInProgress()).isEmpty(); + } + + private DefaultEventPublicationRegistry createRegistry(Instant instant) { + + var clock = Clock.fixed(instant, ZoneId.systemDefault()); + + return new DefaultEventPublicationRegistry(repository, clock); + } + + private Consumer failingConsumer() { + return __ -> { + throw new IllegalStateException(); + }; + } }