Adrastopoulos · Oct 18, 2026
diff --git a/...c/main/java/org/springframework/modulith/events/core/DefaultEventPublicationRegistry.java b/...c/main/java/org/springframework/modulith/events/core/DefaultEventPublicationRegistry.java
@@ -236,7 +236,7 @@ public void processIncompletePublications(Predicate<EventPublication> filter,
 
 		LOGGER.debug(getConfirmationMessage(publications) + " found.");
 
-		processPublications(publications, filter, consumer);
+		processPublications(publications, null, filter, consumer);
 	}
 
 	/*
@@ -262,7 +262,7 @@ public void processFailedPublications(ResubmissionOptions options, Consumer<Targ
 				.withPublicationsPublishedBefore(clock.instant().minus(options.getMinAge()))
 				.withItemsToRead(itemsToRead);
 
-		processPublications(events.findFailedPublications(criteria), options.getFilter(), consumer);
+		processPublications(events.findFailedPublications(criteria), Status.FAILED, options.getFilter(), consumer);
 	}
 
 	/*
@@ -359,14 +359,15 @@ void markFailed(TargetEventPublication publication) {
 		markFailed(publication.getEvent(), publication.getTargetIdentifier());
 	}
 
-	private void processPublications(Collection<TargetEventPublication> publications, Predicate<EventPublication> filter,
+	private void processPublications(Collection<TargetEventPublication> publications,
+			@Nullable Status expectedCurrentStatus, Predicate<EventPublication> filter,
 			Consumer<TargetEventPublication> consumer) {
 
 		publications.stream() //
 				.filter(filter) //
 				.forEach(it -> {
 
-					if (!events.markResubmitted(it.getIdentifier(), clock.instant())) {
+					if (!events.markResubmitted(it.getIdentifier(), clock.instant(), expectedCurrentStatus)) {
 						return;
 					}
 

diff --git a/...re/src/main/java/org/springframework/modulith/events/core/EventPublicationRepository.java b/...re/src/main/java/org/springframework/modulith/events/core/EventPublicationRepository.java
@@ -101,15 +101,22 @@ default void markFailed(UUID identifier) {}
 	 * Marks the {@link org.springframework.modulith.events.EventPublication} with the given identifier as resubmitted.
 	 * Returns {@literal false} in case the update was not successful, which usually means that the instance identified
 	 * might have been claimed for re-submission by another application instance.
+	 * <p>
+	 * If {@code expectedCurrentStatus} is given, the publication is only claimed if it currently has that status or no
+	 * status at all. This guards against claiming a publication that a concurrent resubmission has claimed and moved on
+	 * (e.g. to {@link org.springframework.modulith.events.EventPublication.Status#PROCESSING}) since it was read; used
+	 * when resubmitting previously-read {@link org.springframework.modulith.events.EventPublication.Status#FAILED}
+	 * publications. Pass {@literal null} to claim any incomplete publication not currently resubmitted.
 	 *
 	 * @param identifier must not be {@literal null}.
 	 * @param resubmissionDate must not be {@literal null}.
+	 * @param expectedCurrentStatus can be {@literal null}.
 	 * @return whether the {@link org.springframework.modulith.events.EventPublication} with the given identifier was
 	 *         marked as resubmitted.
 	 * @since 2.0
 	 * @see org.springframework.modulith.events.EventPublication.Status#RESUBMITTED
 	 */
-	default boolean markResubmitted(UUID identifier, Instant resubmissionDate) {
+	default boolean markResubmitted(UUID identifier, Instant resubmissionDate, @Nullable Status expectedCurrentStatus) {
 		return true;
 	}
 

diff --git a/...va/org/springframework/modulith/events/core/DefaultEventPublicationRegistryUnitTests.java b/...va/org/springframework/modulith/events/core/DefaultEventPublicationRegistryUnitTests.java
@@ -136,6 +136,20 @@ void processFailedPublicationsUsesRemainingInFlightWhenLessThanBatchSize() {
 		verify(repository).findFailedPublications(argThat(criteria -> criteria.getMaxItemsToRead() == 50));
 	}
 
+	@Test // GH-1971
+	void claimsFailedPublicationsForResubmissionOnlyIfStillFailed() {
+
+		var identifier = UUID.randomUUID();
+		var publication = mock(TargetEventPublication.class);
+		when(publication.getIdentifier()).thenReturn(identifier);
+
+		when(repository.findFailedPublications(any())).thenReturn(List.of(publication));
+
+		createRegistry(Instant.now()).processFailedPublications(ResubmissionOptions.defaults(), __ -> {});
+
+		verify(repository).markResubmitted(eq(identifier), any(), eq(Status.FAILED));
+	}
+
 	@Test // GH-1764
 	void marksPublicationAsAbandonedWhenAbandonPolicyApplies() {
 

diff --git a/.../main/java/org/springframework/modulith/events/jdbc/JdbcEventPublicationRepositoryV2.java b/.../main/java/org/springframework/modulith/events/jdbc/JdbcEventPublicationRepositoryV2.java
@@ -406,10 +406,14 @@ public void markFailed(UUID identifier) {
 
 	/*
 	 * (non-Javadoc)
-	 * @see org.springframework.modulith.events.core.EventPublicationRepository#markResubmitted(java.util.UUID, java.time.Instant)
+	 * @see org.springframework.modulith.events.core.EventPublicationRepository#markResubmitted(java.util.UUID, java.time.Instant, org.springframework.modulith.events.EventPublication.Status)
 	 */
 	@Override
-	public boolean markResubmitted(UUID identifier, Instant instant) {
+	public boolean markResubmitted(UUID identifier, Instant instant, @Nullable Status expectedCurrentStatus) {
+
+		var guard = expectedCurrentStatus == null
+				? "STATUS != 'RESUBMITTED'"
+				: "STATUS = '%s'".formatted(expectedCurrentStatus.name());
 
 		var sql = asOneLine("""
 				UPDATE %s
@@ -420,8 +424,8 @@ public boolean markResubmitted(UUID identifier, Instant instant) {
 				WHERE
 						ID = ?
 						AND COMPLETION_DATE IS NULL
-						AND (STATUS IS NULL OR STATUS != 'RESUBMITTED')
-				""".formatted(settings.getTable()));
+						AND (STATUS IS NULL OR %s)
+				""".formatted(settings.getTable(), guard));
 
 		var timestamp = Timestamp.from(instant);
 		var id = uuidToDatabase(identifier);

diff --git a/...pringframework/modulith/events/jdbc/JdbcEventPublicationRepositoryV2IntegrationTests.java b/...pringframework/modulith/events/jdbc/JdbcEventPublicationRepositoryV2IntegrationTests.java
@@ -538,8 +538,8 @@ void claimsResubmissionOnce() {
 
 			var now = Instant.now();
 
-			assertThat(repository.markResubmitted(publication.getIdentifier(), now)).isTrue();
-			assertThat(repository.markResubmitted(publication.getIdentifier(), now)).isFalse();
+			assertThat(repository.markResubmitted(publication.getIdentifier(), now, null)).isTrue();
+			assertThat(repository.markResubmitted(publication.getIdentifier(), now, null)).isFalse();
 		}
 
 		@Test // GH-1321
@@ -553,7 +553,7 @@ void countsByStatus() {
 			repository.markFailed(publication.getIdentifier());
 			assertOneByStatus(Status.FAILED);
 
-			repository.markResubmitted(publication.getIdentifier(), Instant.now());
+			repository.markResubmitted(publication.getIdentifier(), Instant.now(), null);
 			assertOneByStatus(Status.RESUBMITTED);
 		}
 
@@ -626,7 +626,7 @@ void doesNotAbandonPublicationThatHasBeenConcurrentlyResubmitted() {
 			var publication = createPublication(event);
 
 			repository.markFailed(publication.getIdentifier());
-			repository.markResubmitted(publication.getIdentifier(), Instant.now());
+			repository.markResubmitted(publication.getIdentifier(), Instant.now(), null);
 
 			assertThat(repository.markAbandoned(publication.getIdentifier(), Instant.now(), Status.FAILED)).isFalse();
 			assertThat(repository.findByStatus(Status.RESUBMITTED))
@@ -635,15 +635,30 @@ void doesNotAbandonPublicationThatHasBeenConcurrentlyResubmitted() {
 			assertThat(repository.findByStatus(Status.ABANDONED)).isEmpty();
 		}
 
-		@Test // GH-1565
+		@Test // GH-1971
+		void doesNotResubmitFailedPublicationThatHasBeenConcurrentlyResubmitted() {
+
+			var publication = createPublication(new TestEvent("first"));
+
+			repository.markFailed(publication.getIdentifier());
+
+			assertThat(repository.markResubmitted(publication.getIdentifier(), Instant.now(), Status.FAILED)).isTrue();
+
+			repository.markProcessing(publication.getIdentifier());
+
+			assertThat(repository.markResubmitted(publication.getIdentifier(), Instant.now(), Status.FAILED)).isFalse();
+			assertOneByStatus(Status.PROCESSING);
+		}
+
+		@Test // GH-1565, GH-1971
 		void resubmitsPublicationWithNullStatusColumn() {
 
 			var event = new TestEvent("first");
 			var publication = createPublication(event);
 
 			setStatusColumnToNull(publication);
 
-			assertThat(repository.markResubmitted(publication.getIdentifier(), Instant.now())).isTrue();
+			assertThat(repository.markResubmitted(publication.getIdentifier(), Instant.now(), Status.FAILED)).isTrue();
 		}
 
 		@Test // GH-1565
@@ -668,7 +683,7 @@ void doesNotResubmitCompletedPublication() {
 
 			repository.markCompleted(publication.getIdentifier(), Instant.now());
 
-			assertThat(repository.markResubmitted(publication.getIdentifier(), Instant.now())).isFalse();
+			assertThat(repository.markResubmitted(publication.getIdentifier(), Instant.now(), null)).isFalse();
 		}
 
 		@Test // GH-1925

diff --git a/.../src/main/java/org/springframework/modulith/events/jpa/JpaEventPublicationRepository.java b/.../src/main/java/org/springframework/modulith/events/jpa/JpaEventPublicationRepository.java
@@ -165,7 +165,7 @@ class JpaEventPublicationRepository implements EventPublicationRepository {
 			       p.lastResubmissionDate = ?1
 			 where p.id = ?2
 			   and p.completionDate is null
-			   and (p.status is null or p.status != org.springframework.modulith.events.EventPublication$Status.RESUBMITTED)
+			   and (p.status is null or %s)
 			""";
 
 	private static final String COUNT_BY_STATUS = """
@@ -374,17 +374,24 @@ public void markFailed(UUID identifier) {
 
 	/*
 	 * (non-Javadoc)
-	 * @see org.springframework.modulith.events.core.EventPublicationRepository#markResubmitted(java.util.UUID, java.time.Instant)
+	 * @see org.springframework.modulith.events.core.EventPublicationRepository#markResubmitted(java.util.UUID, java.time.Instant, org.springframework.modulith.events.EventPublication.Status)
 	 */
 	@Override
-	public boolean markResubmitted(UUID identifier, Instant resubmissionDate) {
+	public boolean markResubmitted(UUID identifier, Instant resubmissionDate, @Nullable Status expectedCurrentStatus) {
 
-		var result = entityManager.createQuery(MARK_RESUBMITTED)
+		var guard = expectedCurrentStatus == null
+				? "p.status != org.springframework.modulith.events.EventPublication$Status.RESUBMITTED"
+				: "p.status = ?3";
+
+		var query = entityManager.createQuery(MARK_RESUBMITTED.formatted(guard))
 				.setParameter(1, resubmissionDate)
-				.setParameter(2, identifier)
-				.executeUpdate();
+				.setParameter(2, identifier);
+
+		if (expectedCurrentStatus != null) {
+			query.setParameter(3, expectedCurrentStatus);
+		}
 
-		return result == 1;
+		return query.executeUpdate() == 1;
 	}
 
 	/*

diff --git a/...rg/springframework/modulith/events/jpa/JpaEventPublicationRepositoryIntegrationTests.java b/...rg/springframework/modulith/events/jpa/JpaEventPublicationRepositoryIntegrationTests.java
@@ -461,8 +461,8 @@ void claimsResubmissionOnce() {
 
 			var now = Instant.now();
 
-			assertThat(repository.markResubmitted(publication.getIdentifier(), now)).isTrue();
-			assertThat(repository.markResubmitted(publication.getIdentifier(), now)).isFalse();
+			assertThat(repository.markResubmitted(publication.getIdentifier(), now, null)).isTrue();
+			assertThat(repository.markResubmitted(publication.getIdentifier(), now, null)).isFalse();
 		}
 
 		@Test // GH-1375
@@ -476,7 +476,7 @@ void countsByStatus() {
 			repository.markFailed(publication.getIdentifier());
 			assertOneByStatus(Status.FAILED);
 
-			repository.markResubmitted(publication.getIdentifier(), Instant.now());
+			repository.markResubmitted(publication.getIdentifier(), Instant.now(), null);
 			assertOneByStatus(Status.RESUBMITTED);
 		}
 
@@ -575,7 +575,7 @@ void doesNotAbandonPublicationThatHasBeenConcurrentlyResubmitted() {
 			var publication = createPublication(event);
 
 			repository.markFailed(publication.getIdentifier());
-			repository.markResubmitted(publication.getIdentifier(), Instant.now());
+			repository.markResubmitted(publication.getIdentifier(), Instant.now(), null);
 
 			em.flush();
 			em.clear();
@@ -591,15 +591,30 @@ void doesNotAbandonPublicationThatHasBeenConcurrentlyResubmitted() {
 			assertThat(repository.findByStatus(Status.ABANDONED)).isEmpty();
 		}
 
-		@Test // GH-1565
+		@Test // GH-1971
+		void doesNotResubmitFailedPublicationThatHasBeenConcurrentlyResubmitted() {
+
+			var publication = createPublication(new TestEvent("first"));
+
+			repository.markFailed(publication.getIdentifier());
+
+			assertThat(repository.markResubmitted(publication.getIdentifier(), Instant.now(), Status.FAILED)).isTrue();
+
+			repository.markProcessing(publication.getIdentifier());
+
+			assertThat(repository.markResubmitted(publication.getIdentifier(), Instant.now(), Status.FAILED)).isFalse();
+			assertOneByStatus(Status.PROCESSING);
+		}
+
+		@Test // GH-1565, GH-1971
 		void resubmitsPublicationWithNullStatusColumn() {
 
 			var event = new TestEvent("first");
 			var publication = createPublication(event);
 
 			setStatusColumnToNull(publication);
 
-			assertThat(repository.markResubmitted(publication.getIdentifier(), Instant.now())).isTrue();
+			assertThat(repository.markResubmitted(publication.getIdentifier(), Instant.now(), Status.FAILED)).isTrue();
 		}
 
 		@Test // GH-1565
@@ -628,7 +643,7 @@ void doesNotResubmitCompletedPublication() {
 			em.flush();
 			em.clear();
 
-			assertThat(repository.markResubmitted(publication.getIdentifier(), Instant.now())).isFalse();
+			assertThat(repository.markResubmitted(publication.getIdentifier(), Instant.now(), null)).isFalse();
 		}
 
 		@Test // GH-1925

diff --git a/...n/java/org/springframework/modulith/events/mongodb/MongoDbEventPublicationRepository.java b/...n/java/org/springframework/modulith/events/mongodb/MongoDbEventPublicationRepository.java
@@ -214,12 +214,15 @@ public void markFailed(UUID identifier) {
 
 	/*
 	 * (non-Javadoc)
-	 * @see org.springframework.modulith.events.core.EventPublicationRepository#markResubmitted(java.util.UUID, java.time.Instant)
+	 * @see org.springframework.modulith.events.core.EventPublicationRepository#markResubmitted(java.util.UUID, java.time.Instant, org.springframework.modulith.events.EventPublication.Status)
 	 */
 	@Override
-	public boolean markResubmitted(UUID identifier, Instant resubmissionDate) {
+	public boolean markResubmitted(UUID identifier, Instant resubmissionDate, @Nullable Status expectedCurrentStatus) {
 
-		var query = query(where(ID).is(identifier).and(STATUS).ne(Status.RESUBMITTED).and(COMPLETION_DATE).isNull());
+		var criteria = where(ID).is(identifier).and(COMPLETION_DATE).isNull().and(STATUS);
+		var query = query(expectedCurrentStatus == null
+				? criteria.ne(Status.RESUBMITTED)
+				: criteria.in(expectedCurrentStatus, null));
 		var update = Update.update(STATUS, Status.RESUBMITTED)
 				.inc(COMPLETION_ATTEMPTS, 1)
 				.set(LAST_RESUBMISSION_DATE, resubmissionDate);