GH-1651 - Improve batching in DefaultEventPublicationRegistry.

Author: odrotbohm

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

@@ -236,9 +236,12 @@ public void processFailedPublications(ResubmissionOptions options, Consumer<Targ
 			return;
 		}
 
+		var remainingHeadroom = options.getMaxInFlight() - currentlyResubmitted;
+		var itemsToRead = Math.min(options.getBatchSize(), remainingHeadroom);
+
 		var criteria = FailedCriteria.ALL
 				.withPublicationsPublishedBefore(clock.instant().minus(options.getMinAge()))
-				.withItemsToRead(Math.min(options.getBatchSize(), options.getBatchSize() - currentlyResubmitted));
+				.withItemsToRead(itemsToRead);
 
 		processPublications(events.findFailedPublications(criteria), options.getFilter(), consumer);
 	}

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

@@ -211,12 +211,17 @@ default int countByStatus(Status status) {
 
 	static class FailedCriteria {
 
-		public static FailedCriteria ALL = new FailedCriteria(-1, null);
+		public static final FailedCriteria ALL = new FailedCriteria(-1, null);
 
 		private final long maxItemsToRead;
 		private final @Nullable Instant publicationDateReference;
 
 		private FailedCriteria(long maxItemsToRead, @Nullable Instant publicationDateReference) {
+
+			Assert.isTrue(maxItemsToRead == -1 || maxItemsToRead > 0,
+					() -> "Maximum number of items to read must be -1 (unlimited) or positive, but was: %d!"
+							.formatted(maxItemsToRead));
+
 			this.maxItemsToRead = maxItemsToRead;
 			this.publicationDateReference = publicationDateReference;
 		}
@@ -237,13 +242,19 @@ public FailedCriteria withPublicationsPublishedBefore(Instant reference) {
 		}
 
 		/**
-		 * The number of {@link org.springframework.modulith.events.EventPublication}s to read. Return -1 to indicate you
-		 * want to read all items.
+		 * The number of {@link org.springframework.modulith.events.EventPublication}s to read. {@code -1} reads all
+		 * matching items (no limit); positive values cap how many rows are read. A limit of {@code 0} is not supported —
+		 * callers that have nothing to read should not invoke {@link #findFailedPublications(FailedCriteria)}.
+		 *
+		 * @return {@code -1} or a positive value
 		 */
 		public long getMaxItemsToRead() {
 			return maxItemsToRead;
 		}
 
+		/**
+		 * @param itemsToRead {@code -1} for no limit, or a positive value
+		 */
 		public FailedCriteria withItemsToRead(long itemsToRead) {
 			return new FailedCriteria(itemsToRead, publicationDateReference);
 		}

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

@@ -23,13 +23,16 @@
 import java.time.Clock;
 import java.time.Instant;
 import java.time.ZoneId;
+import java.util.Collections;
 import java.util.function.Consumer;
 import java.util.stream.Stream;
 
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.modulith.events.EventPublication.Status;
+import org.springframework.modulith.events.ResubmissionOptions;
 
 /**
  * Unit tests for {@link DefaultEventPublicationRegistry}.
@@ -99,6 +102,34 @@ void obtainsCorrectInProgressPublicationForIdenticalEvents() {
 		assertThat(inProgress.getPublication(secondEvent, identifier)).containsSame(second);
 	}
 
+	@Test // GH-1650
+	void processFailedPublicationsCapsReadLimitByMaxInFlightHeadroomNotBatchSize() {
+
+		when(repository.countByStatus(Status.RESUBMITTED)).thenReturn(100);
+		when(repository.findFailedPublications(any())).thenReturn(Collections.emptyList());
+
+		var registry = createRegistry(Instant.now());
+
+		registry.processFailedPublications(
+				ResubmissionOptions.defaults().withMaxInFlight(Integer.MAX_VALUE).withBatchSize(100), __ -> {});
+
+		verify(repository).findFailedPublications(argThat(criteria -> criteria.getMaxItemsToRead() == 100));
+	}
+
+	@Test // GH-1650
+	void processFailedPublicationsUsesRemainingInFlightWhenLessThanBatchSize() {
+
+		when(repository.countByStatus(Status.RESUBMITTED)).thenReturn(100);
+		when(repository.findFailedPublications(any())).thenReturn(Collections.emptyList());
+
+		var registry = createRegistry(Instant.now());
+
+		registry.processFailedPublications(
+				ResubmissionOptions.defaults().withMaxInFlight(150).withBatchSize(100), __ -> {});
+
+		verify(repository).findFailedPublications(argThat(criteria -> criteria.getMaxItemsToRead() == 50));
+	}
+
 	private DefaultEventPublicationRegistry createRegistry(Instant instant) {
 
 		var clock = Clock.fixed(instant, ZoneId.systemDefault());

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

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2026 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 static org.assertj.core.api.Assertions.*;
+import static org.springframework.modulith.events.core.EventPublicationRepository.FailedCriteria.*;
+
+import org.junit.jupiter.api.Test;
+
+/**
+ * Unit tests for {@link EventPublicationRepository.FailedCriteria}.
+ *
+ * @author Oliver Drotbohm
+ */
+class FailedCriteriaUnitTests {
+
+	@Test // GH-1650
+	void acceptsNegativeOneAndPositive() {
+		assertThat(ALL.withItemsToRead(-1).getMaxItemsToRead()).isEqualTo(-1);
+		assertThat(ALL.withItemsToRead(1).getMaxItemsToRead()).isEqualTo(1);
+	}
+
+	@Test // GH-1650
+	void rejectsMaxItemsToReadZero() {
+		assertThatIllegalArgumentException().isThrownBy(() -> ALL.withItemsToRead(0));
+	}
+
+	@Test // GH-1650
+	void rejectsMaxItemsToReadSmallerThanNegativeOne() {
+		assertThatIllegalArgumentException().isThrownBy(() -> ALL.withItemsToRead(-2));
+	}
+}

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

@@ -248,7 +248,9 @@ public List<TargetEventPublication> findFailedPublications(FailedCriteria criter
 			throw new IllegalArgumentException("Number of items to read needs to fit into an integer!");
 		}
 
-		return readMapped(defaultQuery(baseCriteria).limit((int) limit));
+		var query = defaultQuery(baseCriteria);
+
+		return readMapped(limit != -1 ? query.limit((int) limit) : query);
 	}
 
 	/*