GH-796 - Revamped event publication registry and JDBC implementation.

Author: odrotbohm

spring-modulith-events/spring-modulith-events-api/src/main/java/org/springframework/modulith/events/EventPublication.java

@@ -19,6 +19,7 @@
 import java.util.Optional;
 import java.util.UUID;
 
+import org.jspecify.annotations.Nullable;
 import org.springframework.context.ApplicationEvent;
 import org.springframework.context.PayloadApplicationEvent;
 
@@ -100,4 +101,24 @@ default boolean isCompleted() {
 	default int compareTo(EventPublication that) {
 		return this.getPublicationDate().compareTo(that.getPublicationDate());
 	}
+
+	Status getStatus();
+
+	@Nullable
+	Instant getLastResubmissionDate();
+
+	int getCompletionAttempts();
+
+	enum Status {
+
+		PUBLISHED,
+
+		PROCESSING,
+
+		COMPLETED,
+
+		FAILED,
+
+		RESUBMITTED;
+	}
 }

spring-modulith-events/spring-modulith-events-api/src/main/java/org/springframework/modulith/events/FailedEventPublications.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright 2023-2025 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;
+
+/**
+ * All uncompleted event publications.
+ *
+ * @author Oliver Drotbohm
+ * @since 2.0
+ */
+public interface FailedEventPublications {
+
+	void resubmit(ResubmissionOptions options);
+}

spring-modulith-events/spring-modulith-events-api/src/main/java/org/springframework/modulith/events/IncompleteEventPublications.java

@@ -41,4 +41,6 @@ public interface IncompleteEventPublications {
 	 * @param duration must not be {@literal null}.
 	 */
 	void resubmitIncompletePublicationsOlderThan(Duration duration);
+
+	void resubmitIncompletePublications(ResubmissionOptions options);
 }

spring-modulith-events/spring-modulith-events-api/src/main/java/org/springframework/modulith/events/ResubmissionOptions.java

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2025 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;
+
+import java.time.Duration;
+import java.util.function.Predicate;
+
+/**
+ * @author Oliver Drotbohm
+ */
+public class ResubmissionOptions {
+
+	private int maxInFlight = Integer.MAX_VALUE;
+	private int batchSize = 100;
+	private Duration minAge = Duration.ZERO;
+	private Predicate<EventPublication> filter = it -> true;
+
+	public int getMaxInFlight() {
+		return maxInFlight;
+	}
+
+	public int getBatchSize() {
+		return batchSize;
+	}
+
+	public Duration getMinAge() {
+		return minAge;
+	}
+
+	public Predicate<EventPublication> getFilter() {
+		return filter;
+	}
+}

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

@@ -33,6 +33,7 @@
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.autoconfigure.task.TaskExecutionProperties;
 import org.springframework.boot.autoconfigure.task.TaskExecutionProperties.Shutdown;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.Lifecycle;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Import;
@@ -42,6 +43,7 @@
 import org.springframework.modulith.events.core.DefaultEventPublicationRegistry;
 import org.springframework.modulith.events.core.EventPublicationRegistry;
 import org.springframework.modulith.events.core.EventPublicationRepository;
+import org.springframework.modulith.events.core.EventPublicationProperties;
 import org.springframework.modulith.events.support.CompletionRegisteringAdvisor;
 import org.springframework.modulith.events.support.PersistentApplicationEventMulticaster;
 import org.springframework.scheduling.annotation.AbstractAsyncConfiguration;
@@ -56,16 +58,17 @@
  * @author Dmitry Belyaev
  */
 @AutoConfiguration
-@Import(AsyncEnablingConfiguration.class)
+@Import({ AsyncEnablingConfiguration.class, StalenessMonitorConfiguration.class })
+@EnableConfigurationProperties(EventPublicationProperties.class)
 public class EventPublicationAutoConfiguration extends EventPublicationConfiguration {
 
 	@Override
 	@Bean
 	@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
 	@ConditionalOnBean(EventPublicationRepository.class)
 	DefaultEventPublicationRegistry eventPublicationRegistry(EventPublicationRepository repository,
-			ObjectProvider<Clock> clock) {
-		return super.eventPublicationRegistry(repository, clock);
+			ObjectProvider<Clock> clock, ObjectProvider<EventPublicationProperties> properties) {
+		return super.eventPublicationRegistry(repository, clock, properties);
 	}
 
 	@Bean

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

@@ -27,6 +27,7 @@
 import org.springframework.modulith.events.core.DefaultEventPublicationRegistry;
 import org.springframework.modulith.events.core.EventPublicationRegistry;
 import org.springframework.modulith.events.core.EventPublicationRepository;
+import org.springframework.modulith.events.core.EventPublicationProperties;
 import org.springframework.modulith.events.support.CompletionRegisteringAdvisor;
 import org.springframework.modulith.events.support.PersistentApplicationEventMulticaster;
 
@@ -43,8 +44,10 @@ class EventPublicationConfiguration {
 	@Bean
 	@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
 	DefaultEventPublicationRegistry eventPublicationRegistry(EventPublicationRepository repository,
-			ObjectProvider<Clock> clock) {
-		return new DefaultEventPublicationRegistry(repository, clock.getIfAvailable(() -> Clock.systemUTC()));
+			ObjectProvider<Clock> clock, ObjectProvider<EventPublicationProperties> properties) {
+
+		return new DefaultEventPublicationRegistry(repository, clock.getIfAvailable(() -> Clock.systemUTC()),
+				properties.getIfAvailable(() -> EventPublicationProperties.DEFAULTS));
 	}
 
 	@Bean

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

@@ -0,0 +1,75 @@
+/*
+ * Copyright 2025 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.config;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.boot.autoconfigure.AutoConfiguration;
+import org.springframework.modulith.events.core.EventPublicationProperties;
+import org.springframework.modulith.events.core.EventPublicationRegistry;
+import org.springframework.modulith.events.support.EventUtils;
+import org.springframework.scheduling.annotation.SchedulingConfigurer;
+import org.springframework.scheduling.config.ScheduledTaskRegistrar;
+import org.springframework.util.Assert;
+
+/**
+ * Configures a fixed-delay task to mark stale in-flight event publications as failed according to the configuration in
+ * {@link EventPublicationProperties}.
+ *
+ * @author Oliver Drotbohm
+ * @since 2.0
+ */
+@AutoConfiguration
+class StalenessMonitorConfiguration implements SchedulingConfigurer {
+
+	private static final Logger LOGGER = LoggerFactory.getLogger(StalenessMonitorConfiguration.class);
+
+	private final EventPublicationRegistry registry;
+	private final EventPublicationProperties properties;
+
+	/**
+	 * Creates a new {@link StalenessMonitorConfiguration} for the given {@link EventPublicationRegistry} and
+	 * {@link EventPublicationProperties}.
+	 *
+	 * @param registry must not be {@literal null}.
+	 * @param properties must not be {@literal null}.
+	 */
+	StalenessMonitorConfiguration(EventPublicationRegistry registry, EventPublicationProperties properties) {
+
+		Assert.notNull(registry, "EventPublicationRegistry must not be null!");
+		Assert.notNull(properties, "EventPublicationRepositoryProperties must not be null!");
+
+		this.registry = registry;
+		this.properties = properties;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.scheduling.annotation.SchedulingConfigurer#configureTasks(org.springframework.scheduling.config.ScheduledTaskRegistrar)
+	 */
+	@Override
+	public void configureTasks(ScheduledTaskRegistrar taskRegistrar) {
+
+		if (properties.monitorStaleness()) {
+
+			LOGGER.info("Checking for stale event publications every {}.",
+					EventUtils.prettyPrint(properties.getStalenessCheckIntervall()));
+
+			taskRegistrar.addFixedDelayTask(registry::markStalePublicationsFailed,
+					properties.getStalenessCheckIntervall());
+		}
+	}
+}

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

@@ -34,7 +34,10 @@ class DefaultEventPublication implements TargetEventPublication {
 	private final Object event;
 	private final PublicationTargetIdentifier targetIdentifier;
 	private final Instant publicationDate;
+	private final Instant lastResubmissionDate;
+	private final int completionAttempts;
 
+	private Status status;
 	private Optional<Instant> completionDate;
 
 	/**
@@ -55,6 +58,9 @@ class DefaultEventPublication implements TargetEventPublication {
 		this.targetIdentifier = targetIdentifier;
 		this.publicationDate = publicationDate;
 		this.completionDate = Optional.empty();
+		this.status = Status.PUBLISHED;
+		this.lastResubmissionDate = publicationDate;
+		this.completionAttempts = 1;
 	}
 
 	/*
@@ -106,7 +112,36 @@ public Optional<Instant> getCompletionDate() {
 	 */
 	@Override
 	public void markCompleted(Instant instant) {
+
 		this.completionDate = Optional.of(instant);
+		this.status = Status.COMPLETED;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.modulith.events.EventPublication#getStatus()
+	 */
+	@Override
+	public Status getStatus() {
+		return status;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.modulith.events.EventPublication#getLastResubmissionDate()
+	 */
+	@Override
+	public @Nullable Instant getLastResubmissionDate() {
+		return lastResubmissionDate;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.modulith.events.EventPublication#getCompletionAttempts()
+	 */
+	@Override
+	public int getCompletionAttempts() {
+		return completionAttempts;
 	}
 
 	/*
@@ -117,7 +152,7 @@ public void markCompleted(Instant instant) {
 	public String toString() {
 
 		return "DefaultEventPublication [event=" + event + ", targetIdentifier=" + targetIdentifier + ", publicationDate="
-				+ publicationDate + ", completionDate=" + completionDate + "]";
+				+ publicationDate + ", completionDate=" + completionDate + ", status=" + status + "]";
 	}
 
 	/*
@@ -138,7 +173,8 @@ public boolean equals(@Nullable Object obj) {
 		return Objects.equals(this.completionDate, that.completionDate) //
 				&& Objects.equals(this.event, that.event) //
 				&& Objects.equals(this.publicationDate, that.publicationDate) //
-				&& Objects.equals(this.targetIdentifier, that.targetIdentifier);
+				&& Objects.equals(this.targetIdentifier, that.targetIdentifier) //
+				&& Objects.equals(this.status, that.status);
 	}
 
 	/*
@@ -147,6 +183,6 @@ public boolean equals(@Nullable Object obj) {
 	 */
 	@Override
 	public int hashCode() {
-		return Objects.hash(completionDate, event, publicationDate, targetIdentifier);
+		return Objects.hash(completionDate, event, publicationDate, targetIdentifier, status);
 	}
 }