From 464037b20f9fc89437711c7edc7c7d47062f03c3 Mon Sep 17 00:00:00 2001 From: AdzerKI Date: Fri, 9 Oct 2026 20:48:50 +0300 Subject: [PATCH] GH-1906 - Allow resubmitting least recently attempted failed publications first. Failed publications are read oldest first, so publications that keep failing or are skipped by the resubmission filter stay at the head of the queue and, once a full batch of them has piled up, keep all other failed publications from being resubmitted. ResubmissionOptions.withLeastRecentlyAttemptedFirst() now makes the repositories read them by the date of their last resubmission instead, falling back to the publication date for the ones never resubmitted. The default order is unchanged. Signed-off-by: AdzerKI --- .../modulith/events/ResubmissionOptions.java | 35 +++++++++++++++---- .../core/DefaultEventPublicationRegistry.java | 3 +- .../core/EventPublicationRepository.java | 30 +++++++++++++--- ...aultEventPublicationRegistryUnitTests.java | 14 ++++++++ .../JdbcEventPublicationRepositoryV2.java | 4 ++- ...blicationRepositoryV2IntegrationTests.java | 30 +++++++++++++++- .../jpa/JpaEventPublicationRepository.java | 4 ++- ...PublicationRepositoryIntegrationTests.java | 30 +++++++++++++++- .../MongoDbEventPublicationRepository.java | 31 ++++++++++++++++ ...MongoDbEventPublicationRepositoryTest.java | 28 +++++++++++++++ .../Neo4jEventPublicationRepository.java | 7 +++- .../Neo4jEventPublicationRepositoryTest.java | 30 +++++++++++++++- .../antora/modules/ROOT/pages/events.adoc | 6 +++- 13 files changed, 233 insertions(+), 19 deletions(-) diff --git a/spring-modulith-events/spring-modulith-events-api/src/main/java/org/springframework/modulith/events/ResubmissionOptions.java b/spring-modulith-events/spring-modulith-events-api/src/main/java/org/springframework/modulith/events/ResubmissionOptions.java index 35a2a377e..c5e669742 100644 --- a/spring-modulith-events/spring-modulith-events-api/src/main/java/org/springframework/modulith/events/ResubmissionOptions.java +++ b/spring-modulith-events/spring-modulith-events-api/src/main/java/org/springframework/modulith/events/ResubmissionOptions.java @@ -32,8 +32,10 @@ public class ResubmissionOptions { private final int batchSize; private final Duration minAge; private final Predicate filter; + private final boolean leastRecentlyAttemptedFirst; - private ResubmissionOptions(int maxInFlight, int batchSize, Duration minAge, Predicate filter) { + private ResubmissionOptions(int maxInFlight, int batchSize, Duration minAge, Predicate filter, + boolean leastRecentlyAttemptedFirst) { Assert.isTrue(maxInFlight > 0, "Max in flight number must be greater than zero!"); Assert.isTrue(batchSize > 0, "Batch size must be greater than zero!"); @@ -45,16 +47,17 @@ private ResubmissionOptions(int maxInFlight, int batchSize, Duration minAge, Pre this.batchSize = batchSize; this.minAge = minAge; this.filter = filter; + this.leastRecentlyAttemptedFirst = leastRecentlyAttemptedFirst; } /** * 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. + * age, including all {@link EventPublication} instances and reading the oldest publications first. * * @return will never be {@literal null}. */ public static ResubmissionOptions defaults() { - return new ResubmissionOptions(Integer.MAX_VALUE, 100, Duration.ZERO, __ -> true); + return new ResubmissionOptions(Integer.MAX_VALUE, 100, Duration.ZERO, __ -> true, false); } public int getMaxInFlight() { @@ -69,7 +72,7 @@ public int getMaxInFlight() { * @return will never be {@literal null}. */ public ResubmissionOptions withMaxInFlight(int maxInFlight) { - return new ResubmissionOptions(maxInFlight, batchSize, minAge, filter); + return new ResubmissionOptions(maxInFlight, batchSize, minAge, filter, leastRecentlyAttemptedFirst); } public int getBatchSize() { @@ -83,7 +86,7 @@ public int getBatchSize() { * @return will never be {@literal null}. */ public ResubmissionOptions withBatchSize(int batchSize) { - return new ResubmissionOptions(maxInFlight, batchSize, minAge, filter); + return new ResubmissionOptions(maxInFlight, batchSize, minAge, filter, leastRecentlyAttemptedFirst); } public Duration getMinAge() { @@ -97,7 +100,7 @@ public Duration getMinAge() { * @return will never be {@literal null}. */ public ResubmissionOptions withMinAge(Duration minAge) { - return new ResubmissionOptions(maxInFlight, batchSize, minAge, filter); + return new ResubmissionOptions(maxInFlight, batchSize, minAge, filter, leastRecentlyAttemptedFirst); } public Predicate getFilter() { @@ -111,6 +114,24 @@ public Predicate getFilter() { * @return will never be {@literal null}. */ public ResubmissionOptions withFilter(Predicate filter) { - return new ResubmissionOptions(maxInFlight, batchSize, minAge, filter); + return new ResubmissionOptions(maxInFlight, batchSize, minAge, filter, leastRecentlyAttemptedFirst); + } + + public boolean isLeastRecentlyAttemptedFirst() { + return leastRecentlyAttemptedFirst; + } + + /** + * Configures failed {@link EventPublication}s to be read in the order of their last re-submission, falling back to + * their publication date for the ones never resubmitted. A publication resubmitted and failed again moves to the end + * of the queue, so that publications that keep failing or are skipped by the {@link #withFilter(Predicate) filter} + * don't fill every batch read. By default, the oldest publications are read first, which retains their publication + * order. + * + * @return will never be {@literal null}. + * @since 2.2 + */ + public ResubmissionOptions withLeastRecentlyAttemptedFirst() { + return new ResubmissionOptions(maxInFlight, batchSize, minAge, filter, true); } } diff --git a/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/DefaultEventPublicationRegistry.java b/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/DefaultEventPublicationRegistry.java index 90dfcf491..4d2ab154b 100644 --- a/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/DefaultEventPublicationRegistry.java +++ b/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/DefaultEventPublicationRegistry.java @@ -260,7 +260,8 @@ public void processFailedPublications(ResubmissionOptions options, Consumer 0, () -> "Maximum number of items to read must be -1 (unlimited) or positive, but was: %d!" @@ -267,6 +269,7 @@ private FailedCriteria(long maxItemsToRead, @Nullable Instant publicationDateRef this.maxItemsToRead = maxItemsToRead; this.publicationDateReference = publicationDateReference; + this.leastRecentlyAttemptedFirst = leastRecentlyAttemptedFirst; } /** @@ -281,7 +284,7 @@ public Instant getPublicationDateReference() { } public FailedCriteria withPublicationsPublishedBefore(Instant reference) { - return new FailedCriteria(maxItemsToRead, reference); + return new FailedCriteria(maxItemsToRead, reference, leastRecentlyAttemptedFirst); } /** @@ -299,7 +302,26 @@ public long getMaxItemsToRead() { * @param itemsToRead {@code -1} for no limit, or a positive value */ public FailedCriteria withItemsToRead(long itemsToRead) { - return new FailedCriteria(itemsToRead, publicationDateReference); + return new FailedCriteria(itemsToRead, publicationDateReference, leastRecentlyAttemptedFirst); + } + + /** + * Whether to read the {@link org.springframework.modulith.events.EventPublication}s in the order of their last + * re-submission, falling back to their publication date for the ones never resubmitted, instead of the oldest + * ones first. + * + * @since 2.2 + */ + public boolean isLeastRecentlyAttemptedFirst() { + return leastRecentlyAttemptedFirst; + } + + /** + * @param leastRecentlyAttemptedFirst whether to read the publications least recently attempted first. + * @since 2.2 + */ + public FailedCriteria withLeastRecentlyAttemptedFirst(boolean leastRecentlyAttemptedFirst) { + return new FailedCriteria(maxItemsToRead, publicationDateReference, leastRecentlyAttemptedFirst); } } } diff --git a/spring-modulith-events/spring-modulith-events-core/src/test/java/org/springframework/modulith/events/core/DefaultEventPublicationRegistryUnitTests.java b/spring-modulith-events/spring-modulith-events-core/src/test/java/org/springframework/modulith/events/core/DefaultEventPublicationRegistryUnitTests.java index 7712ab8b1..c419006b1 100644 --- a/spring-modulith-events/spring-modulith-events-core/src/test/java/org/springframework/modulith/events/core/DefaultEventPublicationRegistryUnitTests.java +++ b/spring-modulith-events/spring-modulith-events-core/src/test/java/org/springframework/modulith/events/core/DefaultEventPublicationRegistryUnitTests.java @@ -136,6 +136,20 @@ void processFailedPublicationsUsesRemainingInFlightWhenLessThanBatchSize() { verify(repository).findFailedPublications(argThat(criteria -> criteria.getMaxItemsToRead() == 50)); } + @Test // GH-1906 + void processFailedPublicationsReadsLeastRecentlyAttemptedFirstOnlyIfConfigured() { + + when(repository.findFailedPublications(any())).thenReturn(Collections.emptyList()); + + var registry = createRegistry(Instant.now()); + + registry.processFailedPublications(ResubmissionOptions.defaults(), __ -> {}); + registry.processFailedPublications(ResubmissionOptions.defaults().withLeastRecentlyAttemptedFirst(), __ -> {}); + + verify(repository).findFailedPublications(argThat(criteria -> !criteria.isLeastRecentlyAttemptedFirst())); + verify(repository).findFailedPublications(argThat(criteria -> criteria.isLeastRecentlyAttemptedFirst())); + } + @Test // GH-1764 void marksPublicationAsAbandonedWhenAbandonPolicyApplies() { diff --git a/spring-modulith-events/spring-modulith-events-jdbc/src/main/java/org/springframework/modulith/events/jdbc/JdbcEventPublicationRepositoryV2.java b/spring-modulith-events/spring-modulith-events-jdbc/src/main/java/org/springframework/modulith/events/jdbc/JdbcEventPublicationRepositoryV2.java index 90775f448..fafe1a304 100644 --- a/spring-modulith-events/spring-modulith-events-jdbc/src/main/java/org/springframework/modulith/events/jdbc/JdbcEventPublicationRepositoryV2.java +++ b/spring-modulith-events/spring-modulith-events-jdbc/src/main/java/org/springframework/modulith/events/jdbc/JdbcEventPublicationRepositoryV2.java @@ -602,7 +602,9 @@ public List findFailedPublications(FailedCriteria criter args.add(Timestamp.from(instant)); } - sql += " ORDER BY PUBLICATION_DATE ASC"; + sql += criteria.isLeastRecentlyAttemptedFirst() + ? " ORDER BY COALESCE(LAST_RESUBMISSION_DATE, PUBLICATION_DATE) ASC" + : " ORDER BY PUBLICATION_DATE ASC"; var itemsToRead = criteria.getMaxItemsToRead(); diff --git a/spring-modulith-events/spring-modulith-events-jdbc/src/test/java/org/springframework/modulith/events/jdbc/JdbcEventPublicationRepositoryV2IntegrationTests.java b/spring-modulith-events/spring-modulith-events-jdbc/src/test/java/org/springframework/modulith/events/jdbc/JdbcEventPublicationRepositoryV2IntegrationTests.java index d561ea055..6f072d38e 100644 --- a/spring-modulith-events/spring-modulith-events-jdbc/src/test/java/org/springframework/modulith/events/jdbc/JdbcEventPublicationRepositoryV2IntegrationTests.java +++ b/spring-modulith-events/spring-modulith-events-jdbc/src/test/java/org/springframework/modulith/events/jdbc/JdbcEventPublicationRepositoryV2IntegrationTests.java @@ -579,6 +579,30 @@ void looksUpFailedPublicationInBatch() { .containsExactly(publication.getIdentifier()); } + @Test // GH-1906 + void looksUpLeastRecentlyAttemptedFailedPublicationsFirstOnlyIfConfigured() { + + var now = Instant.now(); + var first = createPublication(new TestEvent("first"), now.minusSeconds(2)); + var second = createPublication(new TestEvent("second"), now.minusSeconds(1)); + + repository.markFailed(first.getIdentifier()); + repository.markFailed(second.getIdentifier()); + + repository.markResubmitted(first.getIdentifier(), now); + repository.markFailed(first.getIdentifier()); + + var criteria = FailedCriteria.ALL.withItemsToRead(1); + + assertThat(repository.findFailedPublications(criteria)) + .extracting(TargetEventPublication::getIdentifier) + .containsExactly(first.getIdentifier()); + + assertThat(repository.findFailedPublications(criteria.withLeastRecentlyAttemptedFirst(true))) + .extracting(TargetEventPublication::getIdentifier) + .containsExactly(second.getIdentifier()); + } + @Test // GH-1321 void looksUpFailedPublicationWithReferenceDate() throws Exception { @@ -722,13 +746,17 @@ String archiveTable() { } private TargetEventPublication createPublication(Object event) { + return createPublication(event, Instant.now()); + } + + private TargetEventPublication createPublication(Object event, Instant publicationDate) { var token = event.toString(); doReturn(token).when(serializer).serialize(event); doReturn(event).when(serializer).deserialize(token, event.getClass()); - return repository.create(TargetEventPublication.of(event, TARGET_IDENTIFIER)); + return repository.create(TargetEventPublication.of(event, TARGET_IDENTIFIER, publicationDate)); } } diff --git a/spring-modulith-events/spring-modulith-events-jpa/src/main/java/org/springframework/modulith/events/jpa/JpaEventPublicationRepository.java b/spring-modulith-events/spring-modulith-events-jpa/src/main/java/org/springframework/modulith/events/jpa/JpaEventPublicationRepository.java index 689e3c518..b6c633332 100644 --- a/spring-modulith-events/spring-modulith-events-jpa/src/main/java/org/springframework/modulith/events/jpa/JpaEventPublicationRepository.java +++ b/spring-modulith-events/spring-modulith-events-jpa/src/main/java/org/springframework/modulith/events/jpa/JpaEventPublicationRepository.java @@ -470,7 +470,9 @@ public List findFailedPublications(FailedCriteria criter args.add(instant); } - query += " order by p.publicationDate asc"; + query += criteria.isLeastRecentlyAttemptedFirst() + ? " order by coalesce(p.lastResubmissionDate, p.publicationDate) asc" + : " order by p.publicationDate asc"; var jpaQuery = entityManager.createQuery(query, DefaultJpaEventPublication.class); diff --git a/spring-modulith-events/spring-modulith-events-jpa/src/test/java/org/springframework/modulith/events/jpa/JpaEventPublicationRepositoryIntegrationTests.java b/spring-modulith-events/spring-modulith-events-jpa/src/test/java/org/springframework/modulith/events/jpa/JpaEventPublicationRepositoryIntegrationTests.java index 301cac968..76accba97 100644 --- a/spring-modulith-events/spring-modulith-events-jpa/src/test/java/org/springframework/modulith/events/jpa/JpaEventPublicationRepositoryIntegrationTests.java +++ b/spring-modulith-events/spring-modulith-events-jpa/src/test/java/org/springframework/modulith/events/jpa/JpaEventPublicationRepositoryIntegrationTests.java @@ -502,6 +502,30 @@ void looksUpFailedPublicationInBatch() { .containsExactly(publication.getIdentifier()); } + @Test // GH-1906 + void looksUpLeastRecentlyAttemptedFailedPublicationsFirstOnlyIfConfigured() { + + var now = Instant.now(); + var first = createPublication(new TestEvent("first"), now.minusSeconds(2)); + var second = createPublication(new TestEvent("second"), now.minusSeconds(1)); + + repository.markFailed(first.getIdentifier()); + repository.markFailed(second.getIdentifier()); + + repository.markResubmitted(first.getIdentifier(), now); + repository.markFailed(first.getIdentifier()); + + var criteria = FailedCriteria.ALL.withItemsToRead(1); + + assertThat(repository.findFailedPublications(criteria)) + .extracting(TargetEventPublication::getIdentifier) + .containsExactly(first.getIdentifier()); + + assertThat(repository.findFailedPublications(criteria.withLeastRecentlyAttemptedFirst(true))) + .extracting(TargetEventPublication::getIdentifier) + .containsExactly(second.getIdentifier()); + } + @Test // GH-1375 void looksUpFailedPublicationWithReferenceDate() throws Exception { @@ -682,10 +706,14 @@ private List getArchivedPublications() { } private TargetEventPublication createPublication(Object event) { + return createPublication(event, Instant.now()); + } + + private TargetEventPublication createPublication(Object event, Instant publicationDate) { serializer.registerSerialization(event); - return repository.create(TargetEventPublication.of(event, TARGET_IDENTIFIER)); + return repository.create(TargetEventPublication.of(event, TARGET_IDENTIFIER, publicationDate)); } private void savePublicationAt(LocalDateTime date) { diff --git a/spring-modulith-events/spring-modulith-events-mongodb/src/main/java/org/springframework/modulith/events/mongodb/MongoDbEventPublicationRepository.java b/spring-modulith-events/spring-modulith-events-mongodb/src/main/java/org/springframework/modulith/events/mongodb/MongoDbEventPublicationRepository.java index dcc3fdcfe..ba8f300ad 100644 --- a/spring-modulith-events/spring-modulith-events-mongodb/src/main/java/org/springframework/modulith/events/mongodb/MongoDbEventPublicationRepository.java +++ b/spring-modulith-events/spring-modulith-events-mongodb/src/main/java/org/springframework/modulith/events/mongodb/MongoDbEventPublicationRepository.java @@ -15,10 +15,12 @@ */ package org.springframework.modulith.events.mongodb; +import static org.springframework.data.mongodb.core.aggregation.Aggregation.*; import static org.springframework.data.mongodb.core.query.Criteria.*; import static org.springframework.data.mongodb.core.query.Query.*; import java.time.Instant; +import java.util.ArrayList; import java.util.Collection; import java.util.List; import java.util.Objects; @@ -33,6 +35,8 @@ import org.springframework.data.mongodb.core.BulkOperations.BulkMode; import org.springframework.data.mongodb.core.FindAndModifyOptions; import org.springframework.data.mongodb.core.MongoTemplate; +import org.springframework.data.mongodb.core.aggregation.AggregationOperation; +import org.springframework.data.mongodb.core.aggregation.ConditionalOperators; import org.springframework.data.mongodb.core.aggregation.Fields; import org.springframework.data.mongodb.core.query.Criteria; import org.springframework.data.mongodb.core.query.Query; @@ -63,6 +67,7 @@ class MongoDbEventPublicationRepository implements EventPublicationRepository { private static final String STATUS = "status"; private static final String COMPLETION_ATTEMPTS = "completionAttempts"; private static final String LAST_RESUBMISSION_DATE = "lastResubmissionDate"; + private static final String LAST_ATTEMPT_DATE = "lastAttemptDate"; private static final Sort DEFAULT_SORT = Sort.by(PUBLICATION_DATE).ascending(); @@ -307,6 +312,10 @@ public List findFailedPublications(FailedCriteria criter throw new IllegalArgumentException("Number of items to read needs to fit into an integer!"); } + if (criteria.isLeastRecentlyAttemptedFirst()) { + return readLeastRecentlyAttemptedFirst(baseCriteria, limit); + } + var query = defaultQuery(baseCriteria); return readMapped(limit != -1 ? query.limit((int) limit) : query); @@ -399,6 +408,28 @@ private List readMapped(Query query, String collection) } + private List readLeastRecentlyAttemptedFirst(Criteria criteria, long limit) { + + var lastAttemptDate = ConditionalOperators.ifNull(LAST_RESUBMISSION_DATE).thenValueOf(PUBLICATION_DATE); + + var operations = new ArrayList(List.of( + match(criteria), + addFields().addFieldWithValue(LAST_ATTEMPT_DATE, lastAttemptDate).build(), + sort(Sort.by(LAST_ATTEMPT_DATE).ascending()))); + + if (limit != -1) { + operations.add(limit(limit)); + } + + var aggregation = newAggregation(MongoDbEventPublication.class, operations); + + return mongoTemplate.aggregate(aggregation, collection, MongoDbEventPublication.class) + .getMappedResults() + .stream() + .map(MongoDbEventPublicationRepository::documentToDomain) + .toList(); + } + private Criteria byEventAndListenerId(Object event, PublicationTargetIdentifier identifier) { var eventAsMongoType = mongoTemplate.getConverter().convertToMongoType(event, TypeInformation.OBJECT); diff --git a/spring-modulith-events/spring-modulith-events-mongodb/src/test/java/org/springframework/modulith/events/mongodb/MongoDbEventPublicationRepositoryTest.java b/spring-modulith-events/spring-modulith-events-mongodb/src/test/java/org/springframework/modulith/events/mongodb/MongoDbEventPublicationRepositoryTest.java index 353a4e047..321dff7ff 100644 --- a/spring-modulith-events/spring-modulith-events-mongodb/src/test/java/org/springframework/modulith/events/mongodb/MongoDbEventPublicationRepositoryTest.java +++ b/spring-modulith-events/spring-modulith-events-mongodb/src/test/java/org/springframework/modulith/events/mongodb/MongoDbEventPublicationRepositoryTest.java @@ -563,6 +563,30 @@ void looksUpFailedPublicationInBatch() { .containsExactly(publication.getIdentifier()); } + @Test // GH-1906 + void looksUpLeastRecentlyAttemptedFailedPublicationsFirstOnlyIfConfigured() { + + var now = Instant.now(); + var first = createPublication(new TestEvent("first"), now.minusSeconds(2)); + var second = createPublication(new TestEvent("second"), now.minusSeconds(1)); + + repository.markFailed(first.getIdentifier()); + repository.markFailed(second.getIdentifier()); + + repository.markResubmitted(first.getIdentifier(), now); + repository.markFailed(first.getIdentifier()); + + var criteria = FailedCriteria.ALL.withItemsToRead(1); + + assertThat(repository.findFailedPublications(criteria)) + .extracting(TargetEventPublication::getIdentifier) + .containsExactly(first.getIdentifier()); + + assertThat(repository.findFailedPublications(criteria.withLeastRecentlyAttemptedFirst(true))) + .extracting(TargetEventPublication::getIdentifier) + .containsExactly(second.getIdentifier()); + } + @Test // GH-1321 void looksUpFailedPublicationWithReferenceDate() throws Exception { @@ -682,6 +706,10 @@ private TargetEventPublication createPublication(Object event, PublicationTarget return repository.create(TargetEventPublication.of(event, id)); } + private TargetEventPublication createPublication(Object event, Instant publicationDate) { + return repository.create(TargetEventPublication.of(event, TARGET_IDENTIFIER, publicationDate)); + } + private void savePublicationAt(LocalDateTime date) { var now = date.toInstant(ZoneOffset.UTC); diff --git a/spring-modulith-events/spring-modulith-events-neo4j/src/main/java/org/springframework/modulith/events/neo4j/Neo4jEventPublicationRepository.java b/spring-modulith-events/spring-modulith-events-neo4j/src/main/java/org/springframework/modulith/events/neo4j/Neo4jEventPublicationRepository.java index cf33cb38e..cee32524f 100644 --- a/spring-modulith-events/spring-modulith-events-neo4j/src/main/java/org/springframework/modulith/events/neo4j/Neo4jEventPublicationRepository.java +++ b/spring-modulith-events/spring-modulith-events-neo4j/src/main/java/org/springframework/modulith/events/neo4j/Neo4jEventPublicationRepository.java @@ -663,9 +663,14 @@ public List findFailedPublications(FailedCriteria criter parameters.put(PUBLICATION_DATE, Values.value(instant.atOffset(ZoneOffset.UTC))); } + var publicationDate = EVENT_PUBLICATION_NODE.property(PUBLICATION_DATE); + var order = criteria.isLeastRecentlyAttemptedFirst() + ? coalesce(EVENT_PUBLICATION_NODE.property(LAST_RESUBMISSION_DATE), publicationDate) + : publicationDate; + var limit = criteria.getMaxItemsToRead(); var builder = match.returning(EVENT_PUBLICATION_NODE) - .orderBy(EVENT_PUBLICATION_NODE.property(PUBLICATION_DATE)) + .orderBy(order) .ascending(); var statement = limit != -1 ? builder.limit(limit) : builder; diff --git a/spring-modulith-events/spring-modulith-events-neo4j/src/test/java/org/springframework/modulith/events/neo4j/Neo4jEventPublicationRepositoryTest.java b/spring-modulith-events/spring-modulith-events-neo4j/src/test/java/org/springframework/modulith/events/neo4j/Neo4jEventPublicationRepositoryTest.java index 8ee4adbf7..8db546d81 100644 --- a/spring-modulith-events/spring-modulith-events-neo4j/src/test/java/org/springframework/modulith/events/neo4j/Neo4jEventPublicationRepositoryTest.java +++ b/spring-modulith-events/spring-modulith-events-neo4j/src/test/java/org/springframework/modulith/events/neo4j/Neo4jEventPublicationRepositoryTest.java @@ -431,6 +431,30 @@ void looksUpFailedPublicationInBatch() { .containsExactly(publication.getIdentifier()); } + @Test // GH-1906 + void looksUpLeastRecentlyAttemptedFailedPublicationsFirstOnlyIfConfigured() { + + var now = Instant.now(); + var first = createPublication(new TestEvent("first"), now.minusSeconds(2)); + var second = createPublication(new TestEvent("second"), now.minusSeconds(1)); + + repository.markFailed(first.getIdentifier()); + repository.markFailed(second.getIdentifier()); + + repository.markResubmitted(first.getIdentifier(), now); + repository.markFailed(first.getIdentifier()); + + var criteria = EventPublicationRepository.FailedCriteria.ALL.withItemsToRead(1); + + assertThat(repository.findFailedPublications(criteria)) + .extracting(TargetEventPublication::getIdentifier) + .containsExactly(first.getIdentifier()); + + assertThat(repository.findFailedPublications(criteria.withLeastRecentlyAttemptedFirst(true))) + .extracting(TargetEventPublication::getIdentifier) + .containsExactly(second.getIdentifier()); + } + @Test // GH-1337 void looksUpFailedPublicationWithReferenceDate() throws Exception { @@ -639,13 +663,17 @@ private void assertOneByStatus(EventPublication.Status reference) { } private TargetEventPublication createPublication(Object event) { + return createPublication(event, Instant.now()); + } + + private TargetEventPublication createPublication(Object event, Instant publicationDate) { var token = event.toString(); doReturn(token).when(eventSerializer).serialize(event); doReturn(event).when(eventSerializer).deserialize(token, event.getClass()); - return repository.create(TargetEventPublication.of(event, TARGET_IDENTIFIER)); + return repository.create(TargetEventPublication.of(event, TARGET_IDENTIFIER, publicationDate)); } } diff --git a/src/docs/antora/modules/ROOT/pages/events.adoc b/src/docs/antora/modules/ROOT/pages/events.adoc index 31486aa9b..4bedde262 100644 --- a/src/docs/antora/modules/ROOT/pages/events.adoc +++ b/src/docs/antora/modules/ROOT/pages/events.adoc @@ -279,7 +279,11 @@ If all of `published`, `processing`, and `resubmitted` are zero (default), the S The registry lets you work explicitly with failed publications: * _FailedEventPublications_ (since 2.0) – Use the bean of this type to resubmit only failed publications: `resubmit(ResubmissionOptions)`. -* _ResubmissionOptions_ – Control how resubmission works: batch size, maximum in-flight, minimum age of the publication, and an optional filter (e.g. by event type or `completionAttempts`). Create with `ResubmissionOptions.defaults()` and customize with `withBatchSize(…)`, `withMinAge(…)`, `withFilter(…)`, etc. +* _ResubmissionOptions_ – Control how resubmission works: batch size, maximum in-flight, minimum age of the publication, an optional filter (e.g. by event type or `completionAttempts`), and the order in which failed publications are read. Create with `ResubmissionOptions.defaults()` and customize with `withBatchSize(…)`, `withMinAge(…)`, `withFilter(…)`, `withLeastRecentlyAttemptedFirst()`, etc. + +Failed publications are read oldest first, so that they are resubmitted in the order they were published. +The filter only applies to the publications read, so publications that keep failing or that the filter skips stay at the head of the queue and, once they fill a batch, keep all others from being resubmitted. +If your listeners do not depend on that order, `withLeastRecentlyAttemptedFirst()` reads failed publications in the order of their last resubmission instead, falling back to the publication date for the ones never resubmitted, so that a publication resubmitted and failed again moves to the end of the queue. Resubmission changes the status from `FAILED` to `RESUBMITTED` and updates the last resubmission date; when a listener is about to run, the publication moves to `PROCESSING` and the completion attempt count is incremented.