Make ChangeStreamOperationSpecification more resilient

Remove assumption that expected change stream documents are
returned in a single batch.

JAVA-3742

Author: jyemin

driver-core/src/test/functional/com/mongodb/OperationFunctionalSpecification.groovy

@@ -52,8 +52,8 @@ import com.mongodb.internal.operation.AsyncWriteOperation
 import com.mongodb.internal.operation.InsertOperation
 import com.mongodb.internal.operation.ReadOperation
 import com.mongodb.internal.operation.WriteOperation
-import com.mongodb.internal.validator.NoOpFieldNameValidator
 import com.mongodb.internal.session.SessionContext
+import com.mongodb.internal.validator.NoOpFieldNameValidator
 import org.bson.BsonDocument
 import org.bson.Document
 import org.bson.FieldNameValidator
@@ -156,6 +156,16 @@ class OperationFunctionalSpecification extends Specification {
         results
     }
 
+    def next(cursor, boolean async, int minimumCount) {
+        List<BsonDocument> retVal = []
+
+        while (retVal.size() < minimumCount) {
+            retVal.addAll(next(cursor, async))
+        }
+
+        retVal
+    }
+
     def next(cursor, boolean async) {
         if (async) {
             def futureResultCallback = new FutureResultCallback<List<BsonDocument>>()

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

@@ -21,14 +21,14 @@ import com.mongodb.MongoNamespace
 import com.mongodb.OperationFunctionalSpecification
 import com.mongodb.ReadConcern
 import com.mongodb.WriteConcern
-import com.mongodb.internal.async.SingleResultCallback
 import com.mongodb.client.model.CreateCollectionOptions
 import com.mongodb.client.model.changestream.ChangeStreamDocument
 import com.mongodb.client.model.changestream.FullDocument
 import com.mongodb.client.model.changestream.OperationType
 import com.mongodb.client.model.changestream.UpdateDescription
 import com.mongodb.client.test.CollectionHelper
 import com.mongodb.connection.ConnectionDescription
+import com.mongodb.internal.async.SingleResultCallback
 import com.mongodb.internal.binding.AsyncConnectionSource
 import com.mongodb.internal.binding.AsyncReadBinding
 import com.mongodb.internal.binding.ConnectionSource
@@ -154,7 +154,7 @@ class ChangeStreamOperationSpecification extends OperationFunctionalSpecificatio
         def expected = insertDocuments(helper, [1, 2])
 
         then:
-        def next = nextAndClean(cursor, async)
+        def next = nextAndClean(cursor, async, expected.size())
         next == expected
 
         when:
@@ -163,7 +163,7 @@ class ChangeStreamOperationSpecification extends OperationFunctionalSpecificatio
 
         then:
         cursor.getBatchSize() == 5
-        nextAndClean(cursor, async) == expected
+        nextAndClean(cursor, async, expected.size()) == expected
 
         then:
         if (async) {
@@ -192,7 +192,7 @@ class ChangeStreamOperationSpecification extends OperationFunctionalSpecificatio
         when:
         def cursor = execute(operation, false)
         helper.insertDocuments(BsonDocument.parse('{ _id : 2, x : 2 }'))
-        ChangeStreamDocument<BsonDocument> next = next(cursor, false).get(0)
+        ChangeStreamDocument<BsonDocument> next = next(cursor, false, 1).get(0)
 
         then:
         next.getResumeToken() != null
@@ -219,7 +219,7 @@ class ChangeStreamOperationSpecification extends OperationFunctionalSpecificatio
         when:
         def cursor = execute(operation, false)
         helper.updateOne(BsonDocument.parse('{ _id : 2}'), BsonDocument.parse('{ $set : {x : 3}, $unset : {y : 1}}'))
-        ChangeStreamDocument<BsonDocument> next = next(cursor, false).get(0)
+        ChangeStreamDocument<BsonDocument> next = next(cursor, false, 1).get(0)
 
         then:
         next.getResumeToken() != null
@@ -246,7 +246,7 @@ class ChangeStreamOperationSpecification extends OperationFunctionalSpecificatio
         when:
         def cursor = execute(operation, false)
         helper.replaceOne(BsonDocument.parse('{ _id : 2}'), BsonDocument.parse('{ _id : 2, x : 3}'), false)
-        ChangeStreamDocument<BsonDocument> next = next(cursor, false).get(0)
+        ChangeStreamDocument<BsonDocument> next = next(cursor, false, 1).get(0)
 
         then:
         next.getResumeToken() != null
@@ -273,7 +273,7 @@ class ChangeStreamOperationSpecification extends OperationFunctionalSpecificatio
         when:
         def cursor = execute(operation, false)
         helper.deleteOne(BsonDocument.parse('{ _id : 2}'))
-        ChangeStreamDocument<BsonDocument> next = next(cursor, false).get(0)
+        ChangeStreamDocument<BsonDocument> next = next(cursor, false, 1).get(0)
 
         then:
         next.getResumeToken() != null
@@ -300,7 +300,7 @@ class ChangeStreamOperationSpecification extends OperationFunctionalSpecificatio
         when:
         def cursor = execute(operation, false)
         helper.drop()
-        ChangeStreamDocument<BsonDocument> next = next(cursor, false).get(0)
+        ChangeStreamDocument<BsonDocument> next = next(cursor, false, 1).get(0)
 
         then:
         next.getResumeToken() != null
@@ -328,7 +328,7 @@ class ChangeStreamOperationSpecification extends OperationFunctionalSpecificatio
         when:
         def cursor = execute(operation, false)
         helper.drop()
-        ChangeStreamDocument<BsonDocument> next = next(cursor, false).get(0)
+        ChangeStreamDocument<BsonDocument> next = next(cursor, false, 1).get(0)
 
         then:
         next.getResumeToken() != null
@@ -357,7 +357,7 @@ class ChangeStreamOperationSpecification extends OperationFunctionalSpecificatio
         when:
         def cursor = execute(operation, false)
         helper.dropDatabase('JavaDriverTest')
-        ChangeStreamDocument<BsonDocument> next = next(cursor, false).get(0)
+        ChangeStreamDocument<BsonDocument> next = next(cursor, false, 1).get(0)
 
         then:
         next.getResumeToken() != null
@@ -386,7 +386,7 @@ class ChangeStreamOperationSpecification extends OperationFunctionalSpecificatio
         when:
         def cursor = execute(operation, false)
         helper.renameCollection(newNamespace)
-        ChangeStreamDocument<BsonDocument> next = next(cursor, false).get(0)
+        ChangeStreamDocument<BsonDocument> next = next(cursor, false, 1).get(0)
 
         then:
         next.getResumeToken() != null
@@ -441,7 +441,7 @@ class ChangeStreamOperationSpecification extends OperationFunctionalSpecificatio
         def expected = insertDocuments(helper, [1, 2])
 
         then:
-        nextAndClean(cursor, async) == expected
+        nextAndClean(cursor, async, expected.size()) == expected
 
         then:
         tryNextAndClean(cursor, async) == null
@@ -450,7 +450,7 @@ class ChangeStreamOperationSpecification extends OperationFunctionalSpecificatio
         expected = insertDocuments(helper, [3, 4])
 
         then:
-        nextAndClean(cursor, async) == expected
+        nextAndClean(cursor, async, expected.size()) == expected
 
         cleanup:
         cursor?.close()
@@ -472,15 +472,12 @@ class ChangeStreamOperationSpecification extends OperationFunctionalSpecificatio
         def expected = insertDocuments(helper, [1, 2])
 
         then:
-        nextAndClean(cursor, async) == expected
+        nextAndClean(cursor, async, expected.size()) == expected
 
         when:
         helper.killCursor(helper.getNamespace(), cursor.getWrapped().getServerCursor())
         expected = insertDocuments(helper, [3, 4])
-        def results = nextAndClean(cursor, async)
-        if (results.size() < expected.size()) {
-            results.addAll(nextAndClean(cursor, async))
-        }
+        def results = nextAndClean(cursor, async, expected.size())
 
         then:
         results == expected
@@ -492,10 +489,7 @@ class ChangeStreamOperationSpecification extends OperationFunctionalSpecificatio
         expected = insertDocuments(helper, [5, 6])
         helper.killCursor(helper.getNamespace(), cursor.getWrapped().getServerCursor())
 
-        results = nextAndClean(cursor, async)
-        if (results.size() < expected.size()) {
-            results.addAll(nextAndClean(cursor, async))
-        }
+        results = nextAndClean(cursor, async, expected.size())
 
         then:
         results == expected
@@ -520,7 +514,7 @@ class ChangeStreamOperationSpecification extends OperationFunctionalSpecificatio
 
         when:
         def expected = insertDocuments(helper, [1, 2])
-        def result = next(cursor, async)
+        def result = next(cursor, async, 2)
 
         then:
         result.size() == 2
@@ -531,7 +525,7 @@ class ChangeStreamOperationSpecification extends OperationFunctionalSpecificatio
 
         operation.startAtOperationTime(result.last().getTimestamp('clusterTime'))
         cursor = execute(operation, async)
-        result = nextAndClean(cursor, async)
+        result = nextAndClean(cursor, async, expected.tail.size())
 
         then:
         result == expected.tail()
@@ -555,7 +549,7 @@ class ChangeStreamOperationSpecification extends OperationFunctionalSpecificatio
 
         when:
         def expected = insertDocuments(helper, [1, 2])
-        def result = next(cursor, async)
+        def result = next(cursor, async, 2)
 
         then:
         result.size() == 2
@@ -566,7 +560,7 @@ class ChangeStreamOperationSpecification extends OperationFunctionalSpecificatio
 
         operation.resumeAfter(result.head().getDocument('_id')).startAtOperationTime(null)
         cursor = execute(operation, async)
-        result = nextAndClean(cursor, async)
+        result = nextAndClean(cursor, async, expected.tail().size())
 
         then:
         result == expected.tail()
@@ -591,7 +585,7 @@ class ChangeStreamOperationSpecification extends OperationFunctionalSpecificatio
 
         when:
         def expected = insertDocuments(helper, [1, 2])
-        def result = next(cursor, async)
+        def result = next(cursor, async, 2)
 
         then:
         result.size() == 2
@@ -601,7 +595,7 @@ class ChangeStreamOperationSpecification extends OperationFunctionalSpecificatio
         waitForLastRelease(async ? getAsyncCluster() : getCluster())
 
         cursor = execute(operation.startAfter(result.head().getDocument('_id')).startAtOperationTime(null), async)
-        result = nextAndClean(cursor, async)
+        result = nextAndClean(cursor, async, expected.tail().size())
 
         then:
         result == expected.tail()
@@ -761,6 +755,10 @@ class ChangeStreamOperationSpecification extends OperationFunctionalSpecificatio
         removeExtra(tryNext(cursor, async))
     }
 
+    def nextAndClean(cursor, boolean async, int minimumCount) {
+        removeExtra(next(cursor, async, minimumCount))
+    }
+
     def nextAndClean(cursor, boolean async) {
         removeExtra(next(cursor, async))
     }