GH-1321 - 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;
 
@@ -92,12 +93,69 @@ default boolean isCompleted() {
 		return getCompletionDate().isPresent();
 	}
 
+	/**
+	 * Return the publication's {@link Status}.
+	 *
+	 * @return will never be {@literal null}.
+	 * @since 2.0
+	 */
+	Status getStatus();
+
+	/**
+	 * Returns the last time the {@link EventPublication} was resubmitted.
+	 *
+	 * @return can be {@literal null}.
+	 * @since 2.0
+	 */
+	@Nullable
+	Instant getLastResubmissionDate();
+
+	/**
+	 * Returns the number of completion attempts.
+	 *
+	 * @since 2.0
+	 */
+	int getCompletionAttempts();
+
 	/*
 	 * (non-Javadoc)
 	 * @see java.lang.Comparable#compareTo(java.lang.Object)
 	 */
-	@SuppressWarnings("javadoc")
 	default int compareTo(EventPublication that) {
 		return this.getPublicationDate().compareTo(that.getPublicationDate());
 	}
+
+	/**
+	 * The status of an {@link EventPublication}.
+	 *
+	 * @author Oliver Drotbohm
+	 * @since 2.0
+	 */
+	enum Status {
+
+		/**
+		 * The event publication has been published initially.
+		 */
+		PUBLISHED,
+
+		/**
+		 * An event listener has picked up the publication and is processing it.
+		 */
+		PROCESSING,
+
+		/**
+		 * The processing of the publication has successfully completed.
+		 */
+		COMPLETED,
+
+		/**
+		 * The processing ended up in a failure.
+		 */
+		FAILED,
+
+		/**
+		 * A previously failed publication has been resubmitted for processing.
+		 */
+		RESUBMITTED;
+	}
 }

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

@@ -0,0 +1,32 @@
+/*
+ * 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;
+
+/**
+ * All uncompleted event publications.
+ *
+ * @author Oliver Drotbohm
+ * @since 2.0
+ */
+public interface FailedEventPublications {
+
+	/**
+	 * Initiate the re-submission of failed {@link EventPublication} according to the given {@link ResubmissionOptions}.
+	 *
+	 * @param options must not be {@literal null}.
+	 */
+	void resubmit(ResubmissionOptions options);
+}

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

@@ -41,4 +41,12 @@ public interface IncompleteEventPublications {
 	 * @param duration must not be {@literal null}.
 	 */
 	void resubmitIncompletePublicationsOlderThan(Duration duration);
+
+	/**
+	 * Triggers the re-submission of incomplete {@link EventPublication}s matching the given {@link ResubmissionOptions}.
+	 *
+	 * @param options must not be {@literal null}.
+	 * @since 2.0
+	 */
+	void resubmitIncompletePublications(ResubmissionOptions options);
 }

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

@@ -0,0 +1,116 @@
+/*
+ * 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;
+
+import org.springframework.util.Assert;
+
+/**
+ * Options to be considered during {@link org.springframework.modulith.events.EventPublication} re-submission.
+ *
+ * @author Oliver Drotbohm
+ * @since 2.0
+ */
+public class ResubmissionOptions {
+
+	private final int maxInFlight;
+	private final int batchSize;
+	private final Duration minAge;
+	private final Predicate<EventPublication> filter;
+
+	private ResubmissionOptions(int maxInFlight, int batchSize, Duration minAge, Predicate<EventPublication> filter) {
+
+		Assert.isTrue(maxInFlight > 0, "Max in flight number must be greater than zero!");
+		Assert.isTrue(batchSize > 0, "Batch size must be greater than zero!");
+		Assert.notNull(minAge, "Minimum age must not be null!");
+		Assert.isTrue(!minAge.isNegative(), "Minimum age must not be negative!");
+		Assert.notNull(filter, "Filter must not be null!");
+
+		this.maxInFlight = maxInFlight;
+		this.batchSize = batchSize;
+		this.minAge = minAge;
+		this.filter = filter;
+	}
+
+	/**
+	 * Creates a new {@link ResubmissionOptions} with no bound for in-flight publications, a batch size of 100, no minimum
+	 * age and including all {@link EventPublication} instances.
+	 *
+	 * @return will never be {@literal null}.
+	 */
+	public static ResubmissionOptions defaults() {
+		return new ResubmissionOptions(Integer.MAX_VALUE, 100, Duration.ZERO, __ -> true);
+	}
+
+	public int getMaxInFlight() {
+		return maxInFlight;
+	}
+
+	/**
+	 * Configures the number of publications that are supposed to be in flight concurrently. This means that for each
+	 * re-submission attempt, only a number less than or equal to the configured value will be resubmitted.
+	 *
+	 * @param maxInFlight must not be less than or equal to zero.
+	 * @return will never be {@literal null}.
+	 */
+	public ResubmissionOptions withMaxInFlight(int maxInFlight) {
+		return new ResubmissionOptions(maxInFlight, batchSize, minAge, filter);
+	}
+
+	public int getBatchSize() {
+		return batchSize;
+	}
+
+	/**
+	 * Configures the batch size with which to read publications from the database.
+	 *
+	 * @param batchSize must not be less than or equal to zero.
+	 * @return will never be {@literal null}.
+	 */
+	public ResubmissionOptions withBatchSize(int batchSize) {
+		return new ResubmissionOptions(maxInFlight, batchSize, minAge, filter);
+	}
+
+	public Duration getMinAge() {
+		return minAge;
+	}
+
+	/**
+	 * Configures the minimum age of event publications to qualify for re-submission.
+	 *
+	 * @param minAge must not {@literal null} be negative.
+	 * @return will never be {@literal null}.
+	 */
+	public ResubmissionOptions withMinAge(Duration minAge) {
+		return new ResubmissionOptions(maxInFlight, batchSize, minAge, filter);
+	}
+
+	public Predicate<EventPublication> getFilter() {
+		return filter;
+	}
+
+	/**
+	 * Configures which {@link EventPublication}s to resubmit in a re-submission attempt.
+	 *
+	 * @param filter must not be {@literal null}.
+	 * @return will never be {@literal null}.
+	 */
+	public ResubmissionOptions withFilter(Predicate<EventPublication> filter) {
+		return new ResubmissionOptions(maxInFlight, batchSize, minAge, filter);
+	}
+}

spring-modulith-events/spring-modulith-events-core/pom.xml

@@ -48,6 +48,12 @@
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-autoconfigure</artifactId>
 		</dependency>
+		
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-configuration-processor</artifactId>
+			<optional>true</optional>
+		</dependency>
 
 		<!-- Test -->
 

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;
@@ -56,7 +57,8 @@
  * @author Dmitry Belyaev
  */
 @AutoConfiguration
-@Import(AsyncEnablingConfiguration.class)
+@Import({ AsyncEnablingConfiguration.class, StalenessMonitorConfiguration.class })
+@EnableConfigurationProperties(StalenessProperties.class)
 public class EventPublicationAutoConfiguration extends EventPublicationConfiguration {
 
 	@Override

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

@@ -44,6 +44,7 @@ class EventPublicationConfiguration {
 	@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
 	DefaultEventPublicationRegistry eventPublicationRegistry(EventPublicationRepository repository,
 			ObjectProvider<Clock> clock) {
+
 		return new DefaultEventPublicationRegistry(repository, clock.getIfAvailable(() -> Clock.systemUTC()));
 	}
 

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

@@ -0,0 +1,45 @@
+/*
+ * 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 java.time.Duration;
+
+import org.springframework.util.Assert;
+
+/**
+ * General utilities.
+ *
+ * @author Oliver Drotbohm
+ * @since 2.0
+ */
+class EventUtils {
+
+	/**
+	 * Creates a human-readable {@link String} from a {@link Duration}.
+	 *
+	 * @param duration must not be {@literal null}.
+	 * @return will never be {@literal null}.
+	 */
+	static String prettyPrint(Duration duration) {
+
+		Assert.notNull(duration, "Duration must not be null!");
+
+		return duration.toString()
+				.substring(2)
+				.replaceAll("(\\d[HMS])(?!$)", "$1 ")
+				.toLowerCase();
+	}
+}