GH-258 - Optimize publication completion to issue by-id query.

DefaultEventPublicationRegistry now tracks the event publications currently in progress, so that the completion step can use the database identifier to issue an update statement solely based on that.

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

@@ -17,13 +17,19 @@ package org.springframework.modulith.events.core;
 
 import java.time.Clock;
 import java.time.Duration;
+import java.time.Instant;
 import java.util.Collection;
+import java.util.HashSet;
+import java.util.Optional;
+import java.util.Set;
+import java.util.function.Consumer;
 import java.util.function.Predicate;
 import java.util.stream.Stream;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.DisposableBean;
+import org.springframework.lang.Nullable;
 import org.springframework.modulith.events.CompletedEventPublications;
 import org.springframework.modulith.events.EventPublication;
 import org.springframework.transaction.annotation.Propagation;
@@ -46,6 +52,7 @@ public class DefaultEventPublicationRegistry
 
 	private final EventPublicationRepository events;
 	private final Clock clock;
+	private final PublicationsInProgress inProgress;
 
 	/**
 	 * Creates a new {@link DefaultEventPublicationRegistry} for the given {@link EventPublicationRepository}.
@@ -60,6 +67,7 @@ public class DefaultEventPublicationRegistry
 
 		this.events = events;
 		this.clock = clock;
+		this.inProgress = new PublicationsInProgress();
 	}
 
 	/*
@@ -72,6 +80,7 @@ public class DefaultEventPublicationRegistry
 		return listeners.map(it -> TargetEventPublication.of(event, it, clock.instant()))
 				.peek(it -> LOGGER.debug(REGISTER, it.getEvent().getClass().getName(), it.getTargetIdentifier().getValue()))
 				.map(events::create)
+				.map(inProgress::register)
 				.toList();
 	}
 
@@ -110,7 +119,24 @@ public class DefaultEventPublicationRegistry
 		LOGGER.debug("Marking publication of event {} to listener {} completed.", //
 				event.getClass().getName(), targetIdentifier.getValue());
 
-		events.markCompleted(event, targetIdentifier, clock.instant());
+		Instant now = clock.instant();
+		Runnable fallback = () -> events.markCompleted(event, targetIdentifier, now);
+		Consumer<TargetEventPublication> optimized = it -> {
+			events.markCompleted(it.getIdentifier(), now);
+			inProgress.unregister(event, targetIdentifier);
+		};
+
+		inProgress.getPublication(event, targetIdentifier)
+				.ifPresentOrElse(optimized, fallback);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.modulith.events.core.EventPublicationRegistry#markFailed(java.lang.Object, org.springframework.modulith.events.core.PublicationTargetIdentifier)
+	 */
+	@Override
+	public void markFailed(Object event, PublicationTargetIdentifier targetIdentifier) {
+		inProgress.unregister(event, targetIdentifier);
 	}
 
 	/*
@@ -161,6 +187,42 @@ public class DefaultEventPublicationRegistry
 		events.deleteCompletedPublicationsBefore(now.minus(duration));
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.modulith.events.core.EventPublicationRegistry#processIncompletePublications(java.util.function.Predicate, java.util.function.Consumer, java.time.Duration)
+	 */
+	@Override
+	public void processIncompletePublications(Predicate<EventPublication> filter,
+			Consumer<TargetEventPublication> consumer, @Nullable Duration duration) {
+
+		var message = duration != null ? " older than %s".formatted(duration) : "";
+
+		LOGGER.debug("Looking up incomplete event publications {}… ", message);
+
+		var publications = duration == null //
+				? findIncompletePublications() //
+				: findIncompletePublicationsOlderThan(duration);
+
+		LOGGER.debug(getConfirmationMessage(publications) + " found.");
+
+		publications.stream() //
+				.filter(filter) //
+				.forEach(it -> {
+
+					try {
+
+						inProgress.register(it);
+						consumer.accept(it);
+
+					} catch (Exception o_O) {
+
+						if (LOGGER.isErrorEnabled()) {
+							LOGGER.error("Error republishing event publication " + it, o_O);
+						}
+					}
+				});
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.beans.factory.DisposableBean#destroy()
@@ -186,4 +248,75 @@ public class DefaultEventPublicationRegistry
 			LOGGER.info("{} {} - {}", prefix, it.getEvent().getClass().getName(), it.getTargetIdentifier().getValue());
 		}
 	}
+
+	private static String getConfirmationMessage(Collection<?> publications) {
+
+		var size = publications.size();
+
+		return switch (publications.size()) {
+			case 0 -> "No publication";
+			case 1 -> "1 publication";
+			default -> size + " publications";
+		};
+	}
+
+	/**
+	 * All {@link TargetEventPublication}s currently processed.
+	 *
+	 * @author Oliver Drotbohm
+	 * @since 1.3
+	 */
+	static class PublicationsInProgress {
+
+		private final Set<TargetEventPublication> publications = new HashSet<>();
+
+		/**
+		 * Registers the given {@link TargetEventPublication} as currently processed.
+		 *
+		 * @param publication must not be {@literal null}.
+		 * @return will never be {@literal null}.
+		 */
+		TargetEventPublication register(TargetEventPublication publication) {
+
+			Assert.notNull(publication, "TargetEventPublication must not be null!");
+
+			publications.add(publication);
+
+			return publication;
+		}
+
+		/**
+		 * Unregisters the {@link TargetEventPublication} associated with the given event and
+		 * {@link PublicationTargetIdentifier}.
+		 *
+		 * @param event must not be {@literal null}.
+		 * @param identifier must not be {@literal null}.
+		 */
+		void unregister(Object event, PublicationTargetIdentifier identifier) {
+
+			Assert.notNull(event, "Event must not be null!");
+			Assert.notNull(identifier, "PublicationTargetIdentifier must not be null!");
+
+			getPublication(event, identifier)
+					.ifPresent(publications::remove);
+		}
+
+		/**
+		 * Returns the {@link TargetEventPublication} associated with the given event and
+		 * {@link PublicationTargetIdentifier}.
+		 *
+		 * @param event must not be {@literal null}.
+		 * @param identifier must not be {@literal null}.
+		 * @return will never be {@literal null}.
+		 */
+		Optional<TargetEventPublication> getPublication(Object event, PublicationTargetIdentifier identifier) {
+
+			Assert.notNull(event, "Event must not be null!");
+			Assert.notNull(identifier, "PublicationTargetIdentifier must not be null!");
+
+			return publications.stream()
+					.filter(it -> it.isAssociatedWith(event, identifier))
+					.findFirst();
+		}
+	}
 }

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

@@ -17,9 +17,13 @@ package org.springframework.modulith.events.core;
 
 import java.time.Duration;
 import java.util.Collection;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
 import java.util.stream.Stream;
 
 import org.springframework.context.ApplicationListener;
+import org.springframework.lang.Nullable;
+import org.springframework.modulith.events.EventPublication;
 
 /**
  * A registry to capture event publications to {@link ApplicationListener}s. Allows to register those publications, mark
@@ -64,10 +68,31 @@ public interface EventPublicationRegistry {
 	 */
 	void markCompleted(Object event, PublicationTargetIdentifier targetIdentifier);
 
+	/**
+	 * Marks the publication for the given event and {@link PublicationTargetIdentifier} as failed.
+	 *
+	 * @param event must not be {@literal null}.
+	 * @param targetIdentifier must not be {@literal null}.
+	 * @since 1.3
+	 */
+	void markFailed(Object event, PublicationTargetIdentifier targetIdentifier);
+
 	/**
 	 * Deletes all completed {@link TargetEventPublication}s that have been completed before the given {@link Duration}.
 	 *
 	 * @param duration must not be {@literal null}.
 	 */
 	void deleteCompletedPublicationsOlderThan(Duration duration);
+
+	/**
+	 * Processes all incomplete event publications that have been published before the given duration in relation to "now"
+	 * by applying the given filter passing all remaining instances to the given {@link Consumer}.
+	 *
+	 * @param filter must not be {@literal null}.
+	 * @param consumer must not be {@literal null}.
+	 * @param duration can be {@literal null}.
+	 * @since 1.3
+	 */
+	void processIncompletePublications(Predicate<EventPublication> filter,
+			Consumer<TargetEventPublication> consumer, @Nullable Duration duration);
 }

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

@@ -52,7 +52,7 @@ public interface EventPublicationRepository {
 
 		publication.markCompleted(completionDate);
 
-		markCompleted(publication.getEvent(), publication.getTargetIdentifier(), completionDate);
+		markCompleted(publication.getIdentifier(), completionDate);
 	}
 
 	/**
@@ -65,6 +65,15 @@ public interface EventPublicationRepository {
 	 */
 	void markCompleted(Object event, PublicationTargetIdentifier identifier, Instant completionDate);
 
+	/**
+	 * Marks the publication with the given identifier completed at the given {@link Instant}.
+	 *
+	 * @param identifier must not be {@literal null}.
+	 * @param completionDate must not be {@literal null}.
+	 * @since 1.3
+	 */
+	void markCompleted(UUID identifier, Instant completionDate);
+
 	/**
 	 * Returns all {@link TargetEventPublication}s that have not been completed yet.
 	 *

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

@@ -30,9 +30,9 @@ import org.springframework.util.Assert;
 public interface TargetEventPublication extends Completable, org.springframework.modulith.events.EventPublication {
 
 	/**
-	 * Creates a {@link TargetEventPublication} for the given event an listener identifier using a default {@link Instant}.
-	 * Prefer using {@link #of(Object, PublicationTargetIdentifier, Instant)} with a dedicated {@link Instant} obtained
-	 * from a {@link Clock}.
+	 * Creates a {@link TargetEventPublication} for the given event an listener identifier using a default
+	 * {@link Instant}. Prefer using {@link #of(Object, PublicationTargetIdentifier, Instant)} with a dedicated
+	 * {@link Instant} obtained from a {@link Clock}.
 	 *
 	 * @param event must not be {@literal null}.
 	 * @param id must not be {@literal null}.
@@ -66,7 +66,6 @@ public interface TargetEventPublication extends Completable, org.springframework
 	 * Returns whether the publication is identified by the given {@link PublicationTargetIdentifier}.
 	 *
 	 * @param identifier must not be {@literal null}.
-	 * @return
 	 */
 	default boolean isIdentifiedBy(PublicationTargetIdentifier identifier) {
 
@@ -74,4 +73,20 @@ public interface TargetEventPublication extends Completable, org.springframework
 
 		return this.getTargetIdentifier().equals(identifier);
 	}
+
+	/**
+	 * Returns whether the {@link TargetEventPublication} is associated with the given event and
+	 * {@link PublicationTargetIdentifier}.
+	 *
+	 * @param event must not be {@literal null}.
+	 * @param identifier must not be {@literal null}.
+	 * @since 1.3
+	 */
+	default boolean isAssociatedWith(Object event, PublicationTargetIdentifier identifier) {
+
+		Assert.notNull(event, "Event must not be null!");
+		Assert.notNull(identifier, "PublicationTargetIdentifier must not be null!");
+
+		return isIdentifiedBy(identifier) && getEvent().equals(event);
+	}
 }

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

@@ -32,7 +32,6 @@ 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;
@@ -139,9 +138,8 @@ public class CompletionRegisteringAdvisor extends AbstractPointcutAdvisor {
 	static class CompletionRegisteringMethodInterceptor implements MethodInterceptor, Ordered {
 
 		private static final Logger LOG = LoggerFactory.getLogger(CompletionRegisteringMethodInterceptor.class);
-
-		private static final ConcurrentLruCache<Method, TransactionalApplicationListenerMethodAdapter> ADAPTERS = new ConcurrentLruCache<>(
-				100, CompletionRegisteringMethodInterceptor::createAdapter);
+		private static final ConcurrentLruCache<Method, String> LISTENER_IDS = new ConcurrentLruCache<>(
+				100, CompletionRegisteringMethodInterceptor::lookupListenerId);
 
 		private final @NonNull Supplier<EventPublicationRegistry> registry;
 
@@ -180,14 +178,14 @@ public class CompletionRegisteringAdvisor extends AbstractPointcutAdvisor {
 								return it;
 							})
 							.exceptionallyCompose(it -> {
-								handleFailure(method, it);
+								handleFailure(method, argument, it);
 								return CompletableFuture.failedFuture(it);
 							});
 				}
 
 			} catch (Throwable o_O) {
 
-				handleFailure(method, o_O);
+				handleFailure(method, argument, o_O);
 
 				throw o_O;
 			}
@@ -206,8 +204,9 @@ public class CompletionRegisteringAdvisor extends AbstractPointcutAdvisor {
 			return Ordered.HIGHEST_PRECEDENCE + 10;
 		}
 
-		@Nullable
-		private static Void handleFailure(Method method, Throwable o_O) {
+		private void handleFailure(Method method, Object event, Throwable o_O) {
+
+			markFailed(method, event);
 
 			if (LOG.isDebugEnabled()) {
 				LOG.debug("Invocation of listener {} failed. Leaving event publication uncompleted.", method, o_O);
@@ -215,20 +214,28 @@ public class CompletionRegisteringAdvisor extends AbstractPointcutAdvisor {
 				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();
+			String adapterId = LISTENER_IDS.get(method);
 			PublicationTargetIdentifier identifier = PublicationTargetIdentifier.of(adapterId);
 			registry.get().markCompleted(event, identifier);
 		}
 
-		private static TransactionalApplicationListenerMethodAdapter createAdapter(Method method) {
-			return new TransactionalApplicationListenerMethodAdapter(null, method.getDeclaringClass(), method);
+		private void markFailed(Method method, Object event) {
+
+			// Mark publication complete if the method is a transactional event listener.
+			String adapterId = LISTENER_IDS.get(method);
+			PublicationTargetIdentifier identifier = PublicationTargetIdentifier.of(adapterId);
+			registry.get().markFailed(event, identifier);
+		}
+
+		@SuppressWarnings("null")
+		private static String lookupListenerId(Method method) {
+			return new TransactionalApplicationListenerMethodAdapter(null, method.getDeclaringClass(), method)
+					.getListenerId();
 		}
 	}
 }

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

@@ -196,32 +196,7 @@ public class PersistentApplicationEventMulticaster extends AbstractApplicationEv
 	private void doResubmitUncompletedPublicationsOlderThan(@Nullable Duration duration,
 			Predicate<EventPublication> filter) {
 
-		var message = duration != null ? " older than %s".formatted(duration) : "";
-		var registry = this.registry.get();
-
-		LOGGER.debug("Looking up incomplete event publications {}… ", message);
-
-		var publications = duration == null //
-				? registry.findIncompletePublications() //
-				: registry.findIncompletePublicationsOlderThan(duration);
-
-		LOGGER.debug(getConfirmationMessage(publications) + " found.");
-
-		publications.stream() //
-				.filter(filter) //
-				.forEach(it -> {
-
-					try {
-
-						invokeTargetListener(it);
-
-					} catch (Exception o_O) {
-
-						if (LOGGER.isErrorEnabled()) {
-							LOGGER.error("Error republishing event publication " + it, o_O);
-						}
-					}
-				});
+		registry.get().processIncompletePublications(filter, this::invokeTargetListener, duration);
 	}
 
 	private static ApplicationListener<ApplicationEvent> executeListenerWithCompletion(EventPublication publication,
@@ -281,17 +256,6 @@ public class PersistentApplicationEventMulticaster extends AbstractApplicationEv
 				: (boolean) ReflectionUtils.invokeMethod(LEGACY_SHOULD_HANDLE, candidate, event, new Object[] { payload });
 	}
 
-	private static String getConfirmationMessage(Collection<?> publications) {
-
-		var size = publications.size();
-
-		return switch (publications.size()) {
-			case 0 -> "No publication";
-			case 1 -> "1 publication";
-			default -> size + " publications";
-		};
-	}
-
 	/**
 	 * First-class collection to work with transactional event listeners, i.e. {@link ApplicationListener} instances that
 	 * implement {@link TransactionalApplicationListener}.

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

@@ -0,0 +1,152 @@
+/*
+ * Copyright 2024 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.modulith.events.core;
+
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.function.Predicate;
+
+/**
+ * In-memory implementation of {@link EventPublicationRepository} for testing purposes.
+ *
+ * @author Oliver Drotbohm
+ */
+public class InMemoryEventPublicationRepository
+		implements EventPublicationRepository, Iterable<TargetEventPublication> {
+
+	private static final Predicate<TargetEventPublication> IS_COMPLETED = it -> it.getCompletionDate() != null;
+
+	private Collection<TargetEventPublication> publications = new ArrayList<>();
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.modulith.events.core.EventPublicationRepository#create(org.springframework.modulith.events.core.TargetEventPublication)
+	 */
+	@Override
+	public TargetEventPublication create(TargetEventPublication publication) {
+
+		if (!publications.contains(publication)) {
+			publications.add(publication);
+		}
+
+		return publication;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.modulith.events.core.EventPublicationRepository#markCompleted(java.lang.Object, org.springframework.modulith.events.core.PublicationTargetIdentifier, java.time.Instant)
+	 */
+	@Override
+	public void markCompleted(Object event, PublicationTargetIdentifier identifier, Instant completionDate) {
+
+		publications.stream()
+				.filter(it -> it.isAssociatedWith(event, identifier))
+				.findFirst()
+				.ifPresent(it -> it.markCompleted(completionDate));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.modulith.events.core.EventPublicationRepository#markCompleted(java.util.UUID, java.time.Instant)
+	 */
+	@Override
+	public void markCompleted(UUID identifier, Instant completionDate) {
+
+		publications.stream()
+				.filter(it -> it.getIdentifier().equals(identifier))
+				.findFirst()
+				.ifPresent(it -> it.markCompleted(completionDate));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.modulith.events.core.EventPublicationRepository#findIncompletePublications()
+	 */
+	@Override
+	public List<TargetEventPublication> findIncompletePublications() {
+
+		return publications.stream()
+				.filter(IS_COMPLETED.negate())
+				.toList();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.modulith.events.core.EventPublicationRepository#findIncompletePublicationsPublishedBefore(java.time.Instant)
+	 */
+	@Override
+	public List<TargetEventPublication> findIncompletePublicationsPublishedBefore(Instant instant) {
+
+		return publications.stream()
+				.filter(IS_COMPLETED.negate())
+				.filter(it -> it.getPublicationDate().isBefore(instant))
+				.toList();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.modulith.events.core.EventPublicationRepository#findIncompletePublicationsByEventAndTargetIdentifier(java.lang.Object, org.springframework.modulith.events.core.PublicationTargetIdentifier)
+	 */
+	@Override
+	public Optional<TargetEventPublication> findIncompletePublicationsByEventAndTargetIdentifier(Object event,
+			PublicationTargetIdentifier targetIdentifier) {
+
+		return publications.stream()
+				.filter(it -> it.isAssociatedWith(event, targetIdentifier))
+				.findFirst();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.modulith.events.core.EventPublicationRepository#deletePublications(java.util.List)
+	 */
+	@Override
+	public void deletePublications(List<UUID> identifiers) {
+		publications.removeIf(it -> identifiers.contains(it.getIdentifier()));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.modulith.events.core.EventPublicationRepository#deleteCompletedPublications()
+	 */
+	@Override
+	public void deleteCompletedPublications() {
+		publications.removeIf(IS_COMPLETED);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.modulith.events.core.EventPublicationRepository#deleteCompletedPublicationsBefore(java.time.Instant)
+	 */
+	@Override
+	public void deleteCompletedPublicationsBefore(Instant instant) {
+		publications.removeIf(IS_COMPLETED.and(it -> it.getPublicationDate().isBefore(instant)));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see java.lang.Iterable#iterator()
+	 */
+	@Override
+	public Iterator<TargetEventPublication> iterator() {
+		return publications.iterator();
+	}
+}

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

@@ -15,8 +15,7 @@
  */
 package org.springframework.modulith.events.support;
 
-import static org.mockito.ArgumentMatchers.*;
-import static org.mockito.Mockito.*;
+import static org.assertj.core.api.Assertions.*;
 
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -26,8 +25,8 @@ import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.modulith.events.config.EnablePersistentDomainEvents;
+import org.springframework.modulith.events.core.InMemoryEventPublicationRepository;
 import org.springframework.modulith.events.core.TargetEventPublication;
-import org.springframework.modulith.events.core.EventPublicationRepository;
 import org.springframework.stereotype.Component;
 import org.springframework.test.context.junit.jupiter.SpringExtension;
 import org.springframework.transaction.annotation.EnableTransactionManagement;
@@ -47,8 +46,8 @@ class PersistentApplicationEventMulticasterIntegrationTests {
 	static class TestConfiguration {
 
 		@Bean
-		EventPublicationRepository repository() {
-			return mock(EventPublicationRepository.class);
+		InMemoryEventPublicationRepository repository() {
+			return new InMemoryEventPublicationRepository();
 		}
 
 		@Bean
@@ -58,16 +57,21 @@ class PersistentApplicationEventMulticasterIntegrationTests {
 	}
 
 	@Autowired ApplicationEventPublisher publisher;
-	@Autowired EventPublicationRepository repository;
+	@Autowired InMemoryEventPublicationRepository repository;
 
 	@Test // GH-186, GH-239
 	void doesNotPublishGenericEventsToListeners() throws Exception {
 
 		publisher.publishEvent(new SomeGenericEvent<>());
-		verify(repository, never()).create(any(TargetEventPublication.class));
 
-		publisher.publishEvent(new SomeOtherEvent());
-		verify(repository).create(any(TargetEventPublication.class));
+		assertThat(repository).isEmpty();
+
+		var event = new SomeOtherEvent();
+		publisher.publishEvent(event);
+
+		assertThat(repository)
+				.extracting(TargetEventPublication::getEvent)
+				.containsExactly(event);
 	}
 
 	@Component

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

@@ -16,6 +16,7 @@
 package org.springframework.modulith.events.support;
 
 import static org.assertj.core.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.*;
 import static org.mockito.Mockito.*;
 
 import lombok.AllArgsConstructor;
@@ -72,7 +73,7 @@ class PersistentApplicationEventMulticasterUnitTests {
 
 		multicaster.afterSingletonsInstantiated();
 
-		verify(registry).findIncompletePublications();
+		verify(registry).processIncompletePublications(any(), any(), any());
 	}
 
 	@Test // GH-277

spring-modulith-events/spring-modulith-events-jdbc/src/main/java/org/springframework/modulith/events/jdbc/JdbcEventPublicationRepository.java

@@ -90,6 +90,13 @@ class JdbcEventPublicationRepository implements EventPublicationRepository, Bean
 					AND SERIALIZED_EVENT = ?
 			""";
 
+	private static final String SQL_STATEMENT_UPDATE_BY_ID = """
+			UPDATE %s
+			SET COMPLETION_DATE = ?
+			WHERE
+					ID = ?
+			""";
+
 	private static final String SQL_STATEMENT_FIND_BY_EVENT_AND_LISTENER_ID = """
 			SELECT *
 			FROM %s
@@ -133,6 +140,7 @@ class JdbcEventPublicationRepository implements EventPublicationRepository, Bean
 			sqlStatementFindUncompleted,
 			sqlStatementFindUncompletedBefore,
 			sqlStatementUpdateByEventAndListenerId,
+			sqlStatementUpdateById,
 			sqlStatementFindByEventAndListenerId,
 			sqlStatementDelete,
 			sqlStatementDeleteUncompleted,
@@ -167,6 +175,7 @@ class JdbcEventPublicationRepository implements EventPublicationRepository, Bean
 		this.sqlStatementFindUncompleted = SQL_STATEMENT_FIND_UNCOMPLETED.formatted(table);
 		this.sqlStatementFindUncompletedBefore = SQL_STATEMENT_FIND_UNCOMPLETED_BEFORE.formatted(table);
 		this.sqlStatementUpdateByEventAndListenerId = SQL_STATEMENT_UPDATE_BY_EVENT_AND_LISTENER_ID.formatted(table);
+		this.sqlStatementUpdateById = SQL_STATEMENT_UPDATE_BY_ID.formatted(table);
 		this.sqlStatementFindByEventAndListenerId = SQL_STATEMENT_FIND_BY_EVENT_AND_LISTENER_ID.formatted(table);
 		this.sqlStatementDelete = SQL_STATEMENT_DELETE.formatted(table);
 		this.sqlStatementDeleteUncompleted = SQL_STATEMENT_DELETE_UNCOMPLETED.formatted(table);
@@ -217,6 +226,16 @@ class JdbcEventPublicationRepository implements EventPublicationRepository, Bean
 				serializer.serialize(event));
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.modulith.events.core.EventPublicationRepository#markCompleted(java.util.UUID, java.time.Instant)
+	 */
+	@Override
+	@Transactional
+	public void markCompleted(UUID identifier, Instant completionDate) {
+		operations.update(sqlStatementUpdateById, Timestamp.from(completionDate), uuidToDatabase(identifier));
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.modulith.events.core.EventPublicationRepository#findIncompletePublicationsByEventAndTargetIdentifier(java.lang.Object, org.springframework.modulith.events.core.PublicationTargetIdentifier)

spring-modulith-events/spring-modulith-events-jdbc/src/test/java/org/springframework/modulith/events/jdbc/JdbcEventPublicationRepositoryIntegrationTests.java

@@ -310,6 +310,19 @@ class JdbcEventPublicationRepositoryIntegrationTests {
 					.isEqualTo(event);
 		}
 
+		@Test // GH-258
+		void marksPublicationAsCompletedById() {
+
+			var event = new TestEvent("first");
+			var publication = createPublication(event);
+
+			repository.markCompleted(publication.getIdentifier(), Instant.now());
+
+			assertThat(repository.findCompletedPublications())
+					.extracting(TargetEventPublication::getIdentifier)
+					.containsExactly(publication.getIdentifier());
+		}
+
 		abstract String table();
 
 		private TargetEventPublication createPublication(Object event) {

spring-modulith-events/spring-modulith-events-jpa/src/main/java/org/springframework/modulith/events/jpa/JpaEventPublicationRepository.java

@@ -85,6 +85,12 @@ class JpaEventPublicationRepository implements EventPublicationRepository {
 			   and p.listenerId = ?2
 			""";
 
+	private static final String MARK_COMPLETED_BY_ID = """
+			update JpaEventPublication p
+			   set p.completionDate = ?2
+			 where p.id = ?1
+			""";
+
 	private static final String DELETE = """
 			delete
 			from JpaEventPublication p
@@ -153,6 +159,19 @@ class JpaEventPublicationRepository implements EventPublicationRepository {
 				.executeUpdate();
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.modulith.events.core.EventPublicationRepository#markCompleted(java.util.UUID, java.time.Instant)
+	 */
+	@Override
+	public void markCompleted(UUID identifier, Instant completionDate) {
+
+		entityManager.createQuery(MARK_COMPLETED_BY_ID)
+				.setParameter(1, identifier)
+				.setParameter(2, completionDate)
+				.executeUpdate();
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.modulith.events.EventPublicationRepository#findIncompletePublications()

spring-modulith-events/spring-modulith-events-mongodb/src/main/java/org/springframework/modulith/events/mongodb/MongoDbEventPublicationRepository.java

@@ -92,6 +92,18 @@ class MongoDbEventPublicationRepository implements EventPublicationRepository {
 		mongoTemplate.findAndModify(byEventAndListenerId(event, identifier), update, MongoDbEventPublication.class);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.modulith.events.core.EventPublicationRepository#markCompleted(java.util.UUID, java.time.Instant)
+	 */
+	@Override
+	public void markCompleted(UUID identifier, Instant completionDate) {
+
+		var update = Update.update(COMPLETION_DATE, completionDate);
+
+		mongoTemplate.findAndModify(query(where(ID).is(identifier)), update, MongoDbEventPublication.class);
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.modulith.events.core.EventPublicationRepository#findIncompletePublications()

spring-modulith-events/spring-modulith-events-mongodb/src/test/java/org/springframework/modulith/events/mongodb/MongoDbEventPublicationRepositoryTest.java

@@ -144,6 +144,19 @@ class MongoDbEventPublicationRepositoryTest {
 				.isEqualTo(event);
 	}
 
+	@Test // GH-258
+	void marksPublicationAsCompletedById() {
+
+		var event = new TestEvent("first");
+		var publication = createPublication(event);
+
+		repository.markCompleted(publication.getIdentifier(), Instant.now());
+
+		assertThat(repository.findCompletedPublications())
+				.extracting(TargetEventPublication::getIdentifier)
+				.containsExactly(publication.getIdentifier());
+	}
+
 	private TargetEventPublication createPublication(Object event) {
 		return createPublication(event, TARGET_IDENTIFIER);
 	}

spring-modulith-events/spring-modulith-events-neo4j/src/main/java/org/springframework/modulith/events/neo4j/Neo4jEventPublicationRepository.java

@@ -110,6 +110,11 @@ class Neo4jEventPublicationRepository implements EventPublicationRepository {
 			.set(EVENT_PUBLICATION_NODE.property(COMPLETION_DATE).to(Cypher.parameter(COMPLETION_DATE)))
 			.build();
 
+	private static final Statement COMPLETE_BY_ID_STATEMENT = Cypher.match(EVENT_PUBLICATION_NODE)
+			.where(EVENT_PUBLICATION_NODE.property(ID).eq(Cypher.parameter(ID)))
+			.set(EVENT_PUBLICATION_NODE.property(COMPLETION_DATE).to(Cypher.parameter(COMPLETION_DATE)))
+			.build();
+
 	private static final ResultStatement INCOMPLETE_STATEMENT = Cypher.match(EVENT_PUBLICATION_NODE)
 			.where(EVENT_PUBLICATION_NODE.property(COMPLETION_DATE).isNull())
 			.returning(EVENT_PUBLICATION_NODE)
@@ -185,6 +190,20 @@ class Neo4jEventPublicationRepository implements EventPublicationRepository {
 				.run();
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.modulith.events.core.EventPublicationRepository#markCompleted(java.util.UUID, java.time.Instant)
+	 */
+	@Override
+	@Transactional
+	public void markCompleted(UUID identifier, Instant completionDate) {
+
+		neo4jClient.query(renderer.render(COMPLETE_BY_ID_STATEMENT))
+				.bind(Values.value(identifier.toString())).to(ID)
+				.bind(Values.value(completionDate.atOffset(ZoneOffset.UTC))).to(COMPLETION_DATE)
+				.run();
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.modulith.events.core.EventPublicationRepository#findIncompletePublicationsByEventAndTargetIdentifier(java.lang.Object, org.springframework.modulith.events.core.PublicationTargetIdentifier)

spring-modulith-events/spring-modulith-events-neo4j/src/test/java/org/springframework/modulith/events/neo4j/Neo4jEventPublicationRepositoryTest.java

@@ -235,6 +235,19 @@ class Neo4jEventPublicationRepositoryTest {
 				.isEqualTo(event);
 	}
 
+	@Test // GH-258
+	void marksPublicationAsCompletedById() {
+
+		var event = new TestEvent("first");
+		var publication = createPublication(event);
+
+		repository.markCompleted(publication.getIdentifier(), Instant.now());
+
+		assertThat(repository.findCompletedPublications())
+				.extracting(TargetEventPublication::getIdentifier)
+				.containsExactly(publication.getIdentifier());
+	}
+
 	private TargetEventPublication createPublication(Object event) {
 
 		var token = event.toString();