Test: Race fixes

  * Fix race condition in change stream tests where close can async
    call kill cursors but returns immediately.
  * Fix CommandMessage race where the server monitor may advance the
    global currentId.

Author: rozza

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

@@ -31,6 +31,7 @@ import com.mongodb.binding.ReadBinding
 import com.mongodb.client.model.Collation
 import com.mongodb.client.model.CreateCollectionOptions
 import com.mongodb.client.model.Filters
+import com.mongodb.client.test.CollectionHelper
 import com.mongodb.connection.AsyncConnection
 import com.mongodb.connection.ClusterId
 import com.mongodb.connection.Connection
@@ -195,15 +196,14 @@ class AggregateOperationSpecification extends OperationFunctionalSpecification {
     def 'should support changeStreams'() {
         given:
         def expected = [createExpectedChangeNotification(namespace, 0), createExpectedChangeNotification(namespace, 1)]
-
         def pipeline = ['{$changeStream: {}}', '{$project: {"_id.clusterTime": 0, "_id.uuid": 0}}'].collect { BsonDocument.parse(it) }
         def operation = new AggregateOperation<BsonDocument>(namespace, pipeline, new BsonDocumentCodec())
+        def helper = getCollectionHelper()
 
         when:
-        getCollectionHelper().drop()
-        getCollectionHelper().create(getCollectionHelper().getNamespace().getCollectionName(), new CreateCollectionOptions())
+        helper.create(helper.getNamespace().getCollectionName(), new CreateCollectionOptions())
         def cursor = execute(operation, async)
-        getCollectionHelper().insertDocuments(['{_id: 0, a: 0}', '{_id: 1, a: 1}'].collect { BsonDocument.parse(it) })
+        helper.insertDocuments(['{_id: 0, a: 0}', '{_id: 1, a: 1}'].collect { BsonDocument.parse(it) })
 
         then:
         def next = next(cursor, async).collect { doc ->
@@ -214,6 +214,7 @@ class AggregateOperationSpecification extends OperationFunctionalSpecification {
 
         cleanup:
         cursor?.close()
+        helper?.drop()
 
         where:
         async << [true, false]

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

@@ -133,7 +133,8 @@ class ChangeStreamOperationSpecification extends OperationFunctionalSpecificatio
         }
 
         cleanup:
-        //cursor?.close()
+        cursor?.close()
+        helper?.drop()
 
         where:
         async << [true, false]

driver-core/src/test/unit/com/mongodb/connection/CommandMessageSpecification.groovy

@@ -57,7 +57,7 @@ class CommandMessageSpecification extends Specification {
         def byteBuf = new ByteBufNIO(ByteBuffer.wrap(output.toByteArray()))
         def messageHeader = new MessageHeader(byteBuf, 512)
         messageHeader.opCode == OpCode.OP_MSG.value
-        messageHeader.requestId == RequestMessage.currentGlobalId - 1
+        messageHeader.requestId < RequestMessage.currentGlobalId
         messageHeader.responseTo == 0
 
         def expectedCommandDocument = command.clone()
@@ -167,13 +167,14 @@ class CommandMessageSpecification extends Specification {
 
         then:
         messageHeader.opCode == OpCode.OP_MSG.value
-        messageHeader.requestId == RequestMessage.currentGlobalId - 1
+        messageHeader.requestId < RequestMessage.currentGlobalId
         messageHeader.responseTo == 0
         byteBuf.getInt() == 0
         payload.getPosition() == pos1
         payload.hasAnotherSplit()
 
         when:
+        def initialRequestId = messageHeader.requestId
         payload = payload.getNextSplit()
         message = new CommandMessage(namespace, command, fieldNameValidator, ReadPreference.primary(), messageSettings,
                 false, payload, fieldNameValidator)
@@ -184,7 +185,8 @@ class CommandMessageSpecification extends Specification {
 
         then:
         messageHeader.opCode == OpCode.OP_MSG.value
-        messageHeader.requestId == RequestMessage.currentGlobalId - 1
+        messageHeader.requestId < RequestMessage.currentGlobalId
+        messageHeader.requestId > initialRequestId
         messageHeader.responseTo == 0
         byteBuf.getInt() == 1 << 1
         payload.getPosition() == pos2