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.