Remove locking in CursorReadingTask.getState()

pom.xml

@@ -5,7 +5,7 @@
 
 	<groupId>org.springframework.data</groupId>
 	<artifactId>spring-data-mongodb-parent</artifactId>
-	<version>5.1.0-SNAPSHOT</version>
+	<version>5.1.0-GH-5201-SNAPSHOT</version>
 	<packaging>pom</packaging>
 
 	<name>Spring Data MongoDB</name>

spring-data-mongodb-distribution/pom.xml

@@ -15,7 +15,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-mongodb-parent</artifactId>
-		<version>5.1.0-SNAPSHOT</version>
+		<version>5.1.0-GH-5201-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

spring-data-mongodb/pom.xml

@@ -13,7 +13,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-mongodb-parent</artifactId>
-		<version>5.1.0-SNAPSHOT</version>
+		<version>5.1.0-GH-5201-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/messaging/CursorReadingTask.java

@@ -41,6 +41,8 @@
  */
 abstract class CursorReadingTask<T, R> 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<T, R> 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<T> cursor;
 
@@ -177,7 +179,7 @@ public boolean isLongLived() {
 
 	@Override
 	public State getState() {
-		return lock.execute(() -> state);
+		return state;
 	}
 
 	@Override

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> 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;