GH-891 - DefaultEventPublicationRegistry now properly unregisters in-progress publication on failed resubmission.

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.

Author: odrotbohm

spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/DefaultEventPublicationRegistry.java

@@ -20,8 +20,10 @@
 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 void processIncompletePublications(Predicate<EventPublication> filter,
 
 					} 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 void destroy() {
 		}
 	}
 
+	/**
+	 * 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 @@ private static String getConfirmationMessage(Collection<?> publications) {
 	 * @author Oliver Drotbohm
 	 * @since 1.3
 	 */
-	static class PublicationsInProgress {
+	static class PublicationsInProgress implements Iterable<TargetEventPublication> {
 
-		private final Set<TargetEventPublication> publications = new HashSet<>();
+		private final Set<TargetEventPublication> publications = ConcurrentHashMap.newKeySet();
 
 		/**
 		 * Registers the given {@link TargetEventPublication} as currently processed.
@@ -301,6 +330,18 @@ void unregister(Object event, PublicationTargetIdentifier identifier) {
 					.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 @@ Optional<TargetEventPublication> getPublication(Object event, PublicationTargetI
 					.filter(it -> it.isAssociatedWith(event, identifier))
 					.findFirst();
 		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see java.lang.Iterable#iterator()
+		 */
+		@Override
+		public Iterator<TargetEventPublication> iterator() {
+			return new HashSet<>(publications).iterator();
+		}
 	}
 }

spring-modulith-events/spring-modulith-events-core/src/test/java/org/springframework/modulith/events/core/DefaultEventPublicationRegistryUnitTests.java

@@ -23,6 +23,7 @@
 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,17 +40,15 @@
 class DefaultEventPublicationRegistryUnitTests {
 
 	@Mock EventPublicationRepository repository;
-	@Mock Clock clock;
 
 	@Test // GH-206
 	void usesCustomClockIfConfigured() {
 
 		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 @@ void usesCustomClockIfConfigured() {
 			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<TargetEventPublication> failingConsumer() {
+		return __ -> {
+			throw new IllegalStateException();
+		};
+	}
 }