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.

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<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 @@ 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<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 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<TargetEventPublication> failingConsumer() {
+		return __ -> {
+			throw new IllegalStateException();
+		};
+	}
 }