Clear CommandCursorResult.results after next()/tryNext().

driver-core/src/main/com/mongodb/internal/operation/AsyncCommandBatchCursor.java

@@ -60,6 +60,7 @@
 import static com.mongodb.internal.operation.CommandBatchCursorHelper.getMoreCommandDocument;
 import static com.mongodb.internal.operation.CommandBatchCursorHelper.logCommandCursorResult;
 import static com.mongodb.internal.operation.CommandBatchCursorHelper.translateCommandException;
+import static com.mongodb.internal.operation.CommandCursorResult.withEmptyResults;
 import static java.util.Collections.emptyList;
 
 class AsyncCommandBatchCursor<T> implements AsyncAggregateResponseBatchCursor<T> {
@@ -117,6 +118,7 @@ public void next(final SingleResultCallback<List<T>> callback) {
             }
 
             if (serverCursorIsNull || !batchResults.isEmpty()) {
+                commandCursorResult = withEmptyResults(commandCursorResult);
                 funcCallback.onResult(batchResults, null);
             } else {
                 getMore(localServerCursor, funcCallback);
@@ -206,6 +208,7 @@ private void getMoreLoop(final AsyncConnection connection, final ServerCursor se
                     resourceManager.setServerCursor(nextServerCursor);
                     List<T> nextBatch = commandCursorResult.getResults();
                     if (nextServerCursor == null || !nextBatch.isEmpty()) {
+                        commandCursorResult = withEmptyResults(commandCursorResult);
                         callback.onResult(nextBatch, null);
                         return;
                     }

driver-core/src/main/com/mongodb/internal/operation/BatchCursor.java

@@ -60,7 +60,8 @@ public interface BatchCursor<T> extends Iterator<List<T>>, Closeable {
     boolean hasNext();
 
     /**
-     * Returns the next batch of results.  A tailable cursor will block until another batch exists.
+     * Returns the next batch of results as a mutable list. Modifications to the list will not affect the cursor state.
+     * A tailable cursor will block until another batch exists.
      *
      * @return the next batch of results
      * @throws java.util.NoSuchElementException if no next batch exists
@@ -89,7 +90,8 @@ public interface BatchCursor<T> extends Iterator<List<T>>, Closeable {
     int getBatchSize();
 
     /**
-     * A special {@code next()} case that returns the next batch if available or null.
+     * A special {@code next()} case that returns the next batch as a mutable list if available or null.
+     * Modifications to the list will not affect the cursor state.
      *
      * <p>Tailable cursors are an example where this is useful. A call to {@code tryNext()} may return null, but in the future calling
      * {@code tryNext()} would return a new batch if a document had been added to the capped collection.</p>

driver-core/src/main/com/mongodb/internal/operation/CommandBatchCursor.java

@@ -144,6 +144,7 @@ private List<T> doNext() {
 
         List<T> retVal = nextBatch;
         nextBatch = null;
+        commandCursorResult = CommandCursorResult.withEmptyResults(commandCursorResult);
         return retVal;
     }
 

driver-core/src/main/com/mongodb/internal/operation/CommandCursorResult.java

@@ -23,6 +23,7 @@
 import org.bson.BsonDocument;
 import org.bson.BsonTimestamp;
 
+import java.util.Collections;
 import java.util.List;
 
 import static com.mongodb.assertions.Assertions.isTrue;
@@ -60,6 +61,31 @@ public CommandCursorResult(
         this.postBatchResumeToken = cursorDocument.getDocument(POST_BATCH_RESUME_TOKEN, null);
     }
 
+    private CommandCursorResult(
+            final ServerAddress serverAddress,
+            final List<T> results,
+            final MongoNamespace namespace,
+            final long cursorId,
+            @Nullable final BsonTimestamp operationTime,
+            @Nullable final BsonDocument postBatchResumeToken) {
+        this.serverAddress = serverAddress;
+        this.results = results;
+        this.namespace = namespace;
+        this.cursorId = cursorId;
+        this.operationTime = operationTime;
+        this.postBatchResumeToken = postBatchResumeToken;
+    }
+
+    public static <T> CommandCursorResult<T> withEmptyResults(final CommandCursorResult<T> commandCursorResult) {
+        return new CommandCursorResult<>(
+                commandCursorResult.getServerAddress(),
+                Collections.emptyList(),
+                commandCursorResult.getNamespace(),
+                commandCursorResult.getCursorId(),
+                commandCursorResult.getOperationTime(),
+                commandCursorResult.getPostBatchResumeToken());
+    }
+
     /**
      * Gets the namespace.
      *

driver-sync/src/main/com/mongodb/client/internal/MongoBatchCursorAdapter.java

@@ -97,6 +97,7 @@ public ServerAddress getServerAddress() {
 
     private T getNextInBatch() {
         T nextInBatch = curBatch.get(curPos);
+        curBatch.set(curPos, null);
         if (curPos < curBatch.size() - 1) {
             curPos++;
         } else {

driver-sync/src/main/com/mongodb/client/internal/MongoChangeStreamCursorImpl.java

@@ -112,6 +112,7 @@ public ServerAddress getServerAddress() {
 
     private T getNextInBatch() {
         RawBsonDocument nextInBatch = curBatch.get(curPos);
+        curBatch.set(curPos, null);
         resumeToken = nextInBatch.getDocument("_id");
         if (curPos < curBatch.size() - 1) {
             curPos++;

driver-sync/src/test/unit/com/mongodb/client/gridfs/GridFSFindIterableSpecification.groovy

@@ -130,12 +130,13 @@ class GridFSFindIterableSpecification extends Specification {
                         , null),
         ]
         def cursor = {
+            def batchToReturn = cannedResults.collect();
             Stub(BatchCursor) {
                 def count = 0
                 def results
                 def getResult = {
                     count++
-                    results = count == 1 ? cannedResults : null
+                    results = count == 1 ? batchToReturn : null
                     results
                 }
                 next() >> {

driver-sync/src/test/unit/com/mongodb/client/internal/AggregateIterableSpecification.groovy

@@ -395,12 +395,13 @@ class AggregateIterableSpecification extends Specification {
         given:
         def cannedResults = [new Document('_id', 1), new Document('_id', 2), new Document('_id', 3)]
         def cursor = {
+            def batchToReturn = cannedResults.collect()
             Stub(BatchCursor) {
                 def count = 0
                 def results
                 def getResult = {
                     count++
-                    results = count == 1 ? cannedResults : null
+                    results = count == 1 ? batchToReturn : null
                     results
                 }
                 next() >> {
@@ -591,12 +592,13 @@ class AggregateIterableSpecification extends Specification {
         given:
         def cannedResults = [new Document('_id', 1), new Document('_id', 2), new Document('_id', 3)]
         def cursor = {
+            def batchToReturn = cannedResults.collect()
             Stub(BatchCursor) {
                 def count = 0
                 def results
                 def getResult = {
                     count++
-                    results = count == 1 ? cannedResults : null
+                    results = count == 1 ? batchToReturn : null
                     results
                 }
                 next() >> {

driver-sync/src/test/unit/com/mongodb/client/internal/ChangeStreamIterableSpecification.groovy

@@ -160,8 +160,8 @@ class ChangeStreamIterableSpecification extends Specification {
         def cannedResults = ['{_id: {_data: 1}}', '{_id: {_data: 2}}', '{_id: {_data: 3}}'].collect {
             RawBsonDocument.parse(it)
         }
-        def executor = new TestOperationExecutor([cursor(cannedResults), cursor(cannedResults), cursor(cannedResults),
-                                                  cursor(cannedResults)])
+        def executor = new TestOperationExecutor([cursor(cannedResults.collect()), cursor(cannedResults.collect()),
+                                                  cursor(cannedResults.collect()), cursor(cannedResults.collect())])
         def mongoIterable = new ChangeStreamIterableImpl(null, namespace, codecRegistry, readPreference, readConcern, executor, [],
                 Document, ChangeStreamLevel.COLLECTION, true, TIMEOUT_SETTINGS)
 
@@ -208,8 +208,9 @@ class ChangeStreamIterableSpecification extends Specification {
         given:
         def count = 0
         def cannedResults = ['{_id: { _data: 1}}', '{_id: {_data: 2}}', '{_id: {_data: 3}}'].collect { RawBsonDocument.parse(it) }
-        def executor = new TestOperationExecutor([cursor(cannedResults), cursor(cannedResults), cursor(cannedResults),
-                                                  cursor(cannedResults)])
+        def executor = new TestOperationExecutor([cursor(cannedResults.collect()), cursor(cannedResults.collect()),
+                                                  cursor(cannedResults.collect()), cursor(cannedResults.collect()),
+                                                  cursor(cannedResults.collect())])
         def mongoIterable = new ChangeStreamIterableImpl(null, namespace, codecRegistry, readPreference, readConcern, executor, [],
                 Document, ChangeStreamLevel.COLLECTION, true, TIMEOUT_SETTINGS).withDocumentClass(RawBsonDocument)
 

driver-sync/src/test/unit/com/mongodb/client/internal/DistinctIterableSpecification.groovy

@@ -128,12 +128,13 @@ class DistinctIterableSpecification extends Specification {
         given:
         def cannedResults = [new Document('_id', 1), new Document('_id', 2), new Document('_id', 3)]
         def cursor = {
+            def batchToReturn = cannedResults.collect()
             Stub(BatchCursor) {
                 def count = 0
                 def results
                 def getResult = {
                     count++
-                    results = count == 1 ? cannedResults : null
+                    results = count == 1 ? batchToReturn : null
                     results
                 }
                 next() >> {

driver-sync/src/test/unit/com/mongodb/client/internal/FindIterableSpecification.groovy

@@ -221,11 +221,12 @@ class FindIterableSpecification extends Specification {
         def cannedResults = [new Document('_id', 1), new Document('_id', 2), new Document('_id', 3)]
         def cursor = {
             Stub(BatchCursor) {
+                def batchToReturn = cannedResults.collect()
                 def count = 0
                 def results
                 def getResult = {
                     count++
-                    results = count == 1 ? cannedResults : null
+                    results = count == 1 ? batchToReturn : null
                     results
                 }
                 next() >> {

driver-sync/src/test/unit/com/mongodb/client/internal/ListCollectionsIterableSpecification.groovy

@@ -127,12 +127,13 @@ class ListCollectionsIterableSpecification extends Specification {
         given:
         def cannedResults = [new Document('_id', 1), new Document('_id', 2), new Document('_id', 3)]
         def cursor = {
+            def batchToReturn = cannedResults.collect()
             Stub(BatchCursor) {
                 def count = 0
                 def results
                 def getResult = {
                     count++
-                    results = count == 1 ? cannedResults : null
+                    results = count == 1 ? batchToReturn : null
                     results
                 }
                 next() >> {

driver-sync/src/test/unit/com/mongodb/client/internal/ListDatabasesIterableSpecification.groovy

@@ -83,12 +83,13 @@ class ListDatabasesIterableSpecification extends Specification {
         given:
         def cannedResults = [new Document('_id', 1), new Document('_id', 2), new Document('_id', 3)]
         def cursor = {
+            def batchToReturn = cannedResults.collect()
             Stub(BatchCursor) {
                 def count = 0
                 def results
                 def getResult = {
                     count++
-                    results = count == 1 ? cannedResults : null
+                    results = count == 1 ? batchToReturn : null
                     results
                 }
                 next() >> {

driver-sync/src/test/unit/com/mongodb/client/internal/ListIndexesIterableSpecification.groovy

@@ -105,11 +105,12 @@ class ListIndexesIterableSpecification extends Specification {
         def cannedResults = [new Document('_id', 1), new Document('_id', 2), new Document('_id', 3)]
         def cursor = {
             Stub(BatchCursor) {
+                def batchToReturn = cannedResults.collect()
                 def count = 0
                 def results
                 def getResult = {
                     count++
-                    results = count == 1 ? cannedResults : null
+                    results = count == 1 ? batchToReturn : null
                     results
                 }
                 next() >> {

driver-sync/src/test/unit/com/mongodb/client/internal/MapReduceIterableSpecification.groovy

@@ -255,12 +255,13 @@ class MapReduceIterableSpecification extends Specification {
         given:
         def cannedResults = [new Document('_id', 1), new Document('_id', 2), new Document('_id', 3)]
         def cursor = {
+            def batchToReturn = cannedResults.collect()
             Stub(BatchCursor) {
                 def count = 0
                 def results
                 def getResult = {
                     count++
-                    results = count == 1 ? cannedResults : null
+                    results = count == 1 ? batchToReturn : null
                     results
                 }
                 next() >> {

driver-sync/src/test/unit/com/mongodb/client/internal/MongoBatchCursorAdapterSpecification.groovy

@@ -80,42 +80,46 @@ class MongoBatchCursorAdapterSpecification extends Specification {
 
     def 'should get next from batch cursor'() {
         given:
-        def firstBatch = [new Document('x', 1), new Document('x', 1)]
-        def secondBatch = [new Document('x', 2)]
+        def firstBatchFromBatchCursor = [new Document('x', 1), new Document('x', 1)]
+        def expectedFirstBatch = firstBatchFromBatchCursor.collect()
+        def secondBatchFromBatchCursor = [new Document('x', 2)]
+        def expectedSecondBatch = secondBatchFromBatchCursor.collect()
 
         def batchCursor = Stub(BatchCursor)
 
         batchCursor.hasNext() >>> [true, true, true, true, false]
-        batchCursor.next() >>> [firstBatch, secondBatch]
+        batchCursor.next() >>> [firstBatchFromBatchCursor, secondBatchFromBatchCursor]
 
         def cursor = new MongoBatchCursorAdapter(batchCursor)
 
         expect:
         cursor.hasNext()
-        cursor.next() == firstBatch[0]
+        cursor.next() == expectedFirstBatch[0]
         cursor.hasNext()
-        cursor.next() == firstBatch[1]
+        cursor.next() == expectedFirstBatch[1]
         cursor.hasNext()
-        cursor.next() == secondBatch[0]
+        cursor.next() == expectedSecondBatch[0]
         !cursor.hasNext()
     }
 
     def 'should try next from batch cursor'() {
         given:
-        def firstBatch = [new Document('x', 1), new Document('x', 1)]
-        def secondBatch = [new Document('x', 2)]
+        def firstBatchFromBatchCursor = [new Document('x', 1), new Document('x', 1)]
+        def expectedFirstBatch = firstBatchFromBatchCursor.collect()
+        def secondBatchFromBatchCursor = [new Document('x', 2)]
+        def expectedSecondBatch = secondBatchFromBatchCursor.collect()
 
         def batchCursor = Stub(BatchCursor)
 
-        batchCursor.tryNext() >>> [firstBatch, null, secondBatch, null]
+        batchCursor.tryNext() >>> [firstBatchFromBatchCursor, null, secondBatchFromBatchCursor, null]
 
         def cursor = new MongoBatchCursorAdapter(batchCursor)
 
         expect:
-        cursor.tryNext() == firstBatch[0]
-        cursor.tryNext() == firstBatch[1]
+        cursor.tryNext() == expectedFirstBatch[0]
+        cursor.tryNext() == expectedFirstBatch[1]
         cursor.tryNext() == null
-        cursor.tryNext() == secondBatch[0]
+        cursor.tryNext() == expectedSecondBatch[0]
         cursor.tryNext() == null
     }