Test improvements to reduce race conditions.

Added a sleep after killing a cursor for the
ChangeStreamOperationSpecification.
Updated timeouts for the Sync Adapters to 30 secs to help cope with
replicaset changes.
Added a tryMultipleTimes method to allow multiple testing of a condition
before failing.

JAVA-3535

Author: rozza

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

@@ -488,8 +488,13 @@ class ChangeStreamOperationSpecification extends OperationFunctionalSpecificatio
         expected = insertDocuments(helper, [5, 6])
         helper.killCursor(helper.getNamespace(), cursor.getWrapped().getServerCursor())
 
+        def results = nextAndClean(cursor, async)
+        if (results.size() < expected.size()) {
+            results.addAll(nextAndClean(cursor, async))
+        }
+
         then:
-        nextAndClean(cursor, async) == expected
+        results == expected
 
         cleanup:
         cursor?.close()

driver-reactive-streams/src/test/functional/com/mongodb/reactivestreams/client/TestSubscriber.java

@@ -22,6 +22,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
+import java.util.function.Supplier;
 
 public class TestSubscriber<T> implements Subscriber<T> {
 
@@ -165,7 +166,7 @@ public Subscription getSubscription() {
      * @throws AssertionError if the sequence of items observed does not exactly match {@code items}
      */
     public void assertReceivedOnNext(final List<T> items) {
-        if (getOnNextEvents().size() != items.size()) {
+        if (tryMultipleTimes(2, () -> getOnNextEvents().size() != items.size())) {
             throw new AssertionError("Number of items does not match. Provided: " + items.size() + "  Actual: " + getOnNextEvents().size());
         }
 
@@ -184,6 +185,22 @@ public void assertReceivedOnNext(final List<T> items) {
         }
     }
 
+    private boolean tryMultipleTimes(final int noTimes, final Supplier<Boolean> test) {
+        int counter = noTimes;
+        while (counter > 0){
+            if (test.get()) {
+                return true;
+            }
+            counter--;
+            try {
+                Thread.sleep(500);
+            } catch (InterruptedException e) {
+                return false;
+            }
+        }
+        return false;
+    }
+
     /**
      * Assert that a single terminal event occurred, either {@link #onComplete} or {@link #onError}.
      *

driver-reactive-streams/src/test/functional/com/mongodb/reactivestreams/client/gridfs/GridFSPublisherSpecification.groovy

@@ -39,6 +39,7 @@ import java.nio.ByteBuffer
 import java.nio.channels.Channels
 import java.nio.channels.WritableByteChannel
 import java.security.SecureRandom
+
 import static com.mongodb.client.model.Filters.eq
 import static com.mongodb.client.model.Updates.unset
 import static com.mongodb.reactivestreams.client.Fixture.ObservableSubscriber
@@ -47,6 +48,7 @@ import static com.mongodb.reactivestreams.client.Fixture.getMongoClient
 import static com.mongodb.reactivestreams.client.MongoClients.getDefaultCodecRegistry
 import static com.mongodb.reactivestreams.client.internal.Publishers.publishAndFlatten
 import static java.util.Arrays.asList
+import static java.util.concurrent.TimeUnit.MILLISECONDS
 import static java.util.concurrent.TimeUnit.MINUTES
 import static java.util.concurrent.TimeUnit.SECONDS
 import static org.bson.codecs.configuration.CodecRegistries.fromCodecs
@@ -120,7 +122,7 @@ class GridFSPublisherSpecification extends FunctionalSpecification {
         def options = new GridFSUploadOptions().chunkSizeBytes(chunkSize)
 
         when:
-        def fileId = run(gridFSBucket.&uploadFromPublisher, 'myFile', createPublisher(ByteBuffer.wrap(contentBytes)), options)
+        def fileId = run(MINUTES.toMillis(5), gridFSBucket.&uploadFromPublisher, 'myFile', createPublisher(ByteBuffer.wrap(contentBytes)), options)
 
         then:
         run(filesCollection.&countDocuments) == 1
@@ -488,15 +490,23 @@ class GridFSPublisherSpecification extends FunctionalSpecification {
         }
     }
 
-    def run(operation, ... args) {
-        def result = runAndCollect(operation, args)
+    def run(Closure<?> operation, ... args) {
+        run(MINUTES.toMillis(1), operation, *args)
+    }
+
+    def runAndCollect(Closure<?> operation, ... args) {
+        runAndCollect(MINUTES.toMillis(1), operation, *args)
+    }
+
+    def run(long timeout, Closure<?> operation, ... args) {
+        def result = runAndCollect(timeout, operation, args)
         result != null && !result.isEmpty() ? result.get(0) : result
     }
 
-    def runAndCollect(operation, ... args) {
+    def runAndCollect(long timeout, Closure<?> operation, ... args) {
         def subscriber = new ObservableSubscriber()
         operation.call(args).subscribe(subscriber)
-        subscriber.get(1, MINUTES)
+        subscriber.get(timeout, MILLISECONDS)
     }
 
     byte[] concatByteBuffers(List<ByteBuffer> buffers) {

driver-reactive-streams/src/test/functional/com/mongodb/reactivestreams/client/syncadapter/SingleResultSubscriber.java

@@ -33,7 +33,7 @@ class SingleResultSubscriber<T> implements Subscriber<T> {
     @Nullable
     T get() {
         try {
-            if (!latch.await(5, TimeUnit.SECONDS)) {
+            if (!latch.await(30, TimeUnit.SECONDS)) {
                 throw new MongoTimeoutException("Timeout waiting for single result");
             }
             if (exception != null) {

driver-reactive-streams/src/test/functional/com/mongodb/reactivestreams/client/syncadapter/SyncMongoCursor.java

@@ -64,7 +64,7 @@ public void onComplete() {
             }
         });
         try {
-            if (!latch.await(5, TimeUnit.SECONDS)) {
+            if (!latch.await(30, TimeUnit.SECONDS)) {
                 throw new MongoTimeoutException("Timeout waiting for subscription");
             }
         } catch (InterruptedException e) {
@@ -84,9 +84,9 @@ public boolean hasNext() {
             return true;
         }
         try {
-            Object first = results.pollFirst(5, TimeUnit.SECONDS);
+            Object first = results.pollFirst(30, TimeUnit.SECONDS);
             if (first == null) {
-                throw new MongoTimeoutException("Time out!!!");
+                throw new MongoTimeoutException("Time out waiting for result from cursor");
             } else if (first instanceof Throwable) {
                 throw translateError((Throwable) first);
             } else if (first == COMPLETED) {

driver-scala/src/it/scala/org/mongodb/scala/syncadapter/SyncMongoCursor.scala

@@ -51,7 +51,7 @@ case class SyncMongoCursor[T](val observable: Observable[T]) extends MongoCursor
     }
   })
   try {
-    if (!latch.await(5, TimeUnit.SECONDS)) {
+    if (!latch.await(30, TimeUnit.SECONDS)) {
       throw new MongoTimeoutException("Timeout waiting for subscription")
     }
   } catch {
@@ -68,7 +68,7 @@ case class SyncMongoCursor[T](val observable: Observable[T]) extends MongoCursor
     if (nextResult.isDefined) {
       return true
     }
-    val first = results.pollFirst(5, TimeUnit.SECONDS)
+    val first = results.pollFirst(30, TimeUnit.SECONDS)
     first match {
       case n if n == null                 => throw new MongoTimeoutException("Time out!!!")
       case t if t.isInstanceOf[Throwable] => throw translateError(t.asInstanceOf[Throwable])