JAVA-2603: Wait for all connections pools to have no checked out connections after closing an asynchronous cursor.

Avoid a race condition between the async cursor closing and the check for checked out connections in the common cleanup method

Author: jyemin

driver-core/src/test/functional/com/mongodb/connection/ServerHelper.java

@@ -34,6 +34,14 @@ public static void checkPool(final ServerAddress address) {
         checkPool(address, getAsyncCluster());
     }
 
+    public static void waitForLastRelease(final Cluster cluster) {
+        for (ServerDescription cur : cluster.getCurrentDescription().getServerDescriptions()) {
+            if (cur.isOk()) {
+                waitForLastRelease(cur.getAddress(), cluster);
+            }
+        }
+    }
+
     public static void waitForLastRelease(final ServerAddress address, final Cluster cluster) {
         DefaultServer server = (DefaultServer) cluster.selectServer(new ServerAddressSelector(address));
         DefaultConnectionPool connectionProvider = (DefaultConnectionPool) server.getConnectionPool();

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

@@ -55,12 +55,14 @@ import static com.mongodb.ClusterFixture.collectCursorResults
 import static com.mongodb.ClusterFixture.disableMaxTimeFailPoint
 import static com.mongodb.ClusterFixture.enableMaxTimeFailPoint
 import static com.mongodb.ClusterFixture.executeAsync
+import static com.mongodb.ClusterFixture.getAsyncCluster
 import static com.mongodb.ClusterFixture.getBinding
 import static com.mongodb.ClusterFixture.getCluster
-import static com.mongodb.ClusterFixture.isDiscoverableReplicaSet
 import static com.mongodb.ClusterFixture.isSharded
+import static com.mongodb.ClusterFixture.isStandalone
 import static com.mongodb.ClusterFixture.serverVersionAtLeast
 import static com.mongodb.ExplainVerbosity.QUERY_PLANNER
+import static com.mongodb.connection.ServerHelper.waitForLastRelease
 import static com.mongodb.connection.ServerType.STANDALONE
 import static com.mongodb.operation.QueryOperationHelper.getKeyPattern
 import static com.mongodb.operation.ReadConcernHelper.appendReadConcernToCommand
@@ -191,7 +193,7 @@ class AggregateOperationSpecification extends OperationFunctionalSpecification {
         async << [true, false]
     }
 
-    @IgnoreIf({ !(serverVersionAtLeast(3, 6) && isDiscoverableReplicaSet()) })
+    @IgnoreIf({ !(serverVersionAtLeast(3, 6) && !isStandalone()) })
     def 'should support changeStreams'() {
         given:
         def expected = [createExpectedChangeNotification(namespace, 0), createExpectedChangeNotification(namespace, 1)]
@@ -213,7 +215,7 @@ class AggregateOperationSpecification extends OperationFunctionalSpecification {
 
         cleanup:
         cursor?.close()
-        helper?.drop()
+        waitForLastRelease(async ? getAsyncCluster() : getCluster())
 
         where:
         async << [true, false]

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

@@ -38,8 +38,11 @@ import org.bson.codecs.DocumentCodec
 import org.bson.codecs.ValueCodecProvider
 import spock.lang.IgnoreIf
 
+import static com.mongodb.ClusterFixture.getAsyncCluster
+import static com.mongodb.ClusterFixture.getCluster
 import static com.mongodb.ClusterFixture.isStandalone
 import static com.mongodb.ClusterFixture.serverVersionAtLeast
+import static com.mongodb.connection.ServerHelper.waitForLastRelease
 import static java.util.concurrent.TimeUnit.MILLISECONDS
 import static org.bson.codecs.configuration.CodecRegistries.fromProviders
 
@@ -134,7 +137,7 @@ class ChangeStreamOperationSpecification extends OperationFunctionalSpecificatio
 
         cleanup:
         cursor?.close()
-        helper?.drop()
+        waitForLastRelease(async ? getAsyncCluster() : getCluster())
 
         where:
         async << [true, false]
@@ -302,7 +305,7 @@ class ChangeStreamOperationSpecification extends OperationFunctionalSpecificatio
 
         cleanup:
         cursor?.close()
-        helper?.drop()
+        waitForLastRelease(async ? getAsyncCluster() : getCluster())
 
         where:
         async << [true, false]
@@ -341,7 +344,7 @@ class ChangeStreamOperationSpecification extends OperationFunctionalSpecificatio
 
         cleanup:
         cursor?.close()
-        helper?.drop()
+        waitForLastRelease(async ? getAsyncCluster() : getCluster())
 
         where:
         async << [true, false]
@@ -364,6 +367,7 @@ class ChangeStreamOperationSpecification extends OperationFunctionalSpecificatio
 
         when:
         cursor.close()
+        waitForLastRelease(async ? getAsyncCluster() : getCluster())
 
         operation.resumeAfter(result.head().getDocument('_id'))
         cursor = execute(operation, async)
@@ -374,7 +378,7 @@ class ChangeStreamOperationSpecification extends OperationFunctionalSpecificatio
 
         cleanup:
         cursor?.close()
-        helper?.drop()
+        waitForLastRelease(async ? getAsyncCluster() : getCluster())
 
         where:
         async << [true, false]