JAVA-2651: Ensure that cursor iteration eventually stops when the cursor is closed

Author: jyemin

driver-core/src/main/com/mongodb/operation/QueryBatchCursor.java

@@ -57,7 +57,7 @@ class QueryBatchCursor<T> implements BatchCursor<T> {
     private ServerCursor serverCursor;
     private List<T> nextBatch;
     private int count;
-    private boolean closed;
+    private volatile boolean closed;
 
     QueryBatchCursor(final QueryResult<T> firstQueryResult, final int limit, final int batchSize, final Decoder<T> decoder) {
         this(firstQueryResult, limit, batchSize, decoder, null);
@@ -107,6 +107,9 @@ public boolean hasNext() {
 
         while (serverCursor != null) {
             getMore();
+            if (closed) {
+                throw new IllegalStateException("Cursor has been closed");
+            }
             if (nextBatch != null) {
                 return true;
             }

driver-core/src/test/functional/com/mongodb/operation/QueryBatchCursorFunctionalSpecification.groovy

@@ -257,6 +257,27 @@ class QueryBatchCursorFunctionalSpecification extends OperationFunctionalSpecifi
         nextBatch.iterator().next().get('_id') == 2
     }
 
+    @Category(Slow)
+    def 'hasNext should throw when cursor is closed in another thread'() {
+        collectionHelper.create(collectionName, new CreateCollectionOptions().capped(true).sizeInBytes(1000))
+        collectionHelper.insertDocuments(new DocumentCodec(), new Document('_id', 1).append('ts', new BsonTimestamp(5, 0)))
+        def firstBatch = executeQuery(new BsonDocument('ts', new BsonDocument('$gte', new BsonTimestamp(5, 0))), 0, 2, true, true);
+        cursor = new QueryBatchCursor<Document>(firstBatch, 0, 2, new DocumentCodec(), connectionSource)
+        cursor.next()
+
+        // wait a second then close the cursor
+        new Thread({
+            sleep(1000)
+            cursor.close()
+        } as Runnable).start()
+
+        when:
+        cursor.hasNext()
+
+        then:
+        thrown(Exception)
+    }
+
     @IgnoreIf({ !serverVersionAtLeast(3, 2) || isSharded() })
     @Category(Slow)
     def 'test maxTimeMS'() {