From 6ff1ecc850503ea37700a3d3a172a6c9abec7878 Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Fri, 5 Jun 2026 11:04:53 +0200 Subject: [PATCH 1/2] Prepare issue branch. --- pom.xml | 2 +- spring-data-mongodb-distribution/pom.xml | 2 +- spring-data-mongodb/pom.xml | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/pom.xml b/pom.xml index 351c02d827..bc7dc78d8f 100644 --- a/pom.xml +++ b/pom.xml @@ -5,7 +5,7 @@ org.springframework.data spring-data-mongodb-parent - 5.1.0-SNAPSHOT + 5.1.0-GH-5201-SNAPSHOT pom Spring Data MongoDB diff --git a/spring-data-mongodb-distribution/pom.xml b/spring-data-mongodb-distribution/pom.xml index cab02fe276..4e117bdaed 100644 --- a/spring-data-mongodb-distribution/pom.xml +++ b/spring-data-mongodb-distribution/pom.xml @@ -15,7 +15,7 @@ org.springframework.data spring-data-mongodb-parent - 5.1.0-SNAPSHOT + 5.1.0-GH-5201-SNAPSHOT ../pom.xml diff --git a/spring-data-mongodb/pom.xml b/spring-data-mongodb/pom.xml index 0a758111af..a40f9e5ccc 100644 --- a/spring-data-mongodb/pom.xml +++ b/spring-data-mongodb/pom.xml @@ -13,7 +13,7 @@ org.springframework.data spring-data-mongodb-parent - 5.1.0-SNAPSHOT + 5.1.0-GH-5201-SNAPSHOT ../pom.xml From fe6039d0e0c721edd536db31cb77f05abb53226e Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Fri, 5 Jun 2026 11:10:41 +0200 Subject: [PATCH 2/2] Remove locking in `CursorReadingTask.getState()` CursorReadingTask used a single lock to guard both the lifecycle state and access to the underlying MongoCursor. As the cursor read happens through a long-polling tryNext() call that can block for up to maxAwaitTime, any concurrent getState() invocation (e.g. via Subscription.isActive()) had to wait for the in-flight poll to complete before it could obtain the lock, contradicting the non-blocking nature of the operation. We now hold the state in a volatile field and read it without acquiring the lock so that getState() returns immediately regardless of an ongoing cursor read. Lifecycle transitions remain guarded by the lock to keep state changes and cursor handling consistent. Closes #5201 --- .../core/messaging/CursorReadingTask.java | 6 ++- .../messaging/CursorReadingTaskUnitTests.java | 38 +++++++++++++++++++ 2 files changed, 42 insertions(+), 2 deletions(-) diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/messaging/CursorReadingTask.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/messaging/CursorReadingTask.java index ce7bb13fa6..59cdaa1766 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/messaging/CursorReadingTask.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/messaging/CursorReadingTask.java @@ -41,6 +41,8 @@ */ abstract class CursorReadingTask implements Task { + // ensure happens-before ordering to avoid asynchronous operations leak while + // synchronous operations such as close suggest a completed state. private final Lock lock = Lock.of(new ReentrantLock()); private final MongoTemplate template; @@ -49,7 +51,7 @@ abstract class CursorReadingTask implements Task { private final ErrorHandler errorHandler; private final CountDownLatch awaitStart = new CountDownLatch(1); - private State state = State.CREATED; + private volatile State state = State.CREATED; private @Nullable MongoCursor cursor; @@ -177,7 +179,7 @@ public boolean isLongLived() { @Override public State getState() { - return lock.execute(() -> state); + return state; } @Override diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/messaging/CursorReadingTaskUnitTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/messaging/CursorReadingTaskUnitTests.java index 2f41447c22..a40fde12e8 100644 --- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/messaging/CursorReadingTaskUnitTests.java +++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/messaging/CursorReadingTaskUnitTests.java @@ -21,8 +21,12 @@ import edu.umd.cs.mtc.MultithreadedTestCase; +import java.time.Duration; import java.util.List; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -123,6 +127,40 @@ public void errorOnNextNotifiesErrorHandlerOnlyOnce() { assertThat(errorCaptor.getValue()).isInstanceOf(IllegalStateException.class); } + @Test // GH-5201 + public void getStateDoesNotBlockWhileReadingFromCursor() throws Exception { + + when(cursor.getServerCursor()).thenReturn(new ServerCursor(10, new ServerAddress("mock"))); + + CountDownLatch tryNextEntered = new CountDownLatch(1); + CountDownLatch releaseTryNext = new CountDownLatch(1); + + when(cursor.tryNext()).thenAnswer(invocation -> { + + tryNextEntered.countDown(); + assertThat(releaseTryNext.await(5, TimeUnit.SECONDS)).isTrue(); + return null; + }); + + Thread taskThread = new Thread(task::run, "cursor-reading-task-test"); + taskThread.start(); + + try { + + assertThat(task.awaitStart(Duration.ofSeconds(5))).isTrue(); + assertThat(tryNextEntered.await(5, TimeUnit.SECONDS)).isTrue(); + + CompletableFuture state = CompletableFuture.supplyAsync(task::getState); + + assertThat(state.get(1, TimeUnit.SECONDS)).isEqualTo(State.RUNNING); + } finally { + + releaseTryNext.countDown(); + task.cancel(); + taskThread.join(5000); + } + } + private static class MultithreadedStopRunningWhileEmittingMessages extends MultithreadedTestCase { CursorReadingTask task;