Backport to 4.2.x JAVA-4044/PR#689 and the changes it depends on (JAVA-3938 & JAVA-3907 / PR#661) (#691)

* Regression test for change stream cancellation (#661)

Ensures that all sessions are returned to the pool

JAVA-3938 JAVA-3907

* Guarantee that ChangeStreamPublisher for a collection completes after dropping the collection (#689)

Before the changes made within JAVA-3973,
ChangeStreamPublisher had been terminating with onError.
After the changes in JAVA-3973 neither onError nor onComplete is called,
but those changes allow us to terminated it with onComplete.
I could have specified only assertTerminalEvent()
without specifying assertNoErrors(), thus accepting either onComplete or onError
(the old behavior), but terminating with onComplete is nicer.

The approach with using startAtOperationTime to ensure that
a change stream is guaranteed to observe collection.drop()
works only if there is no leader re-election that results in
rolling back the delete operation from which the operationTime
was extracted. While such rollback can be prevented by using
the "majority" write concern, the common approach in driver tests
is to not use it for efficiency and tolerate a tiny chance of
experiencing a rollback.

JAVA-4044

Co-authored-by: Ross Lawley <ross.lawley@gmail.com>

Author: stIncMale

driver-core/src/main/com/mongodb/assertions/Assertions.java

@@ -18,11 +18,18 @@
 package com.mongodb.assertions;
 
 import com.mongodb.internal.async.SingleResultCallback;
+import com.mongodb.lang.Nullable;
 
 import java.util.Collection;
 
 /**
  * <p>Design by contract assertions.</p> <p>This class is not part of the public API and may be removed or changed at any time.</p>
+ * All {@code assert...} methods throw {@link AssertionError} and should be used to check conditions which may be violated if and only if
+ * the driver code is incorrect. The intended usage of this methods is the same as of the
+ * <a href="https://docs.oracle.com/javase/8/docs/technotes/guides/language/assert.html">Java {@code assert} statement</a>. The reason
+ * for not using the {@code assert} statements is that they are not always enabled. We prefer having internal checks always done at the
+ * cost of our code doing a relatively small amount of additional work in production.
+ * The {@code assert...} methods return values to open possibilities of being used fluently.
  */
 public final class Assertions {
     /**
@@ -119,6 +126,57 @@ public static void doesNotContainNull(final String name, final Collection<?> col
         }
     }
 
+    /**
+     * @param value A value to check.
+     * @param <T> The type of {@code value}.
+     * @return {@code null}.
+     * @throws AssertionError If {@code value} is not {@code null}.
+     */
+    @Nullable
+    public static <T> T assertNull(@Nullable final T value) throws AssertionError {
+        if (value != null) {
+            throw new AssertionError(value.toString());
+        }
+        return null;
+    }
+
+    /**
+     * @param value A value to check.
+     * @param <T> The type of {@code value}.
+     * @return {@code value}
+     * @throws AssertionError If {@code value} is {@code null}.
+     */
+    public static <T> T assertNotNull(@Nullable final T value) throws AssertionError {
+        if (value == null) {
+            throw new AssertionError();
+        }
+        return value;
+    }
+
+    /**
+     * @param value A value to check.
+     * @return {@code true}.
+     * @throws AssertionError If {@code value} is {@code false}.
+     */
+    public static boolean assertTrue(final boolean value) throws AssertionError {
+        if (!value) {
+            throw new AssertionError();
+        }
+        return true;
+    }
+
+    /**
+     * @param value A value to check.
+     * @return {@code false}.
+     * @throws AssertionError If {@code value} is {@code true}.
+     */
+    public static boolean assertFalse(final boolean value) throws AssertionError {
+        if (value) {
+            throw new AssertionError();
+        }
+        return false;
+    }
+
     private Assertions() {
     }
 }

driver-core/src/main/com/mongodb/internal/async/AsyncBatchCursor.java

@@ -65,12 +65,24 @@ public interface AsyncBatchCursor<T> extends Closeable {
     int getBatchSize();
 
     /**
-     * Return true if the AsyncBatchCursor has been closed
+     * Implementations of {@link AsyncBatchCursor} are allowed to close themselves, see {@link #close()} for more details.
      *
-     * @return true if the AsyncBatchCursor has been closed
+     * @return {@code true} if {@code this} has been closed or has closed itself.
      */
     boolean isClosed();
 
+    /**
+     * Implementations of {@link AsyncBatchCursor} are allowed to close themselves synchronously via methods
+     * {@link #next(SingleResultCallback)}/{@link #tryNext(SingleResultCallback)}.
+     * Self-closing behavior is discouraged because it introduces an additional burden on code that uses {@link AsyncBatchCursor}.
+     * To help making such code simpler, this method is required to be idempotent.
+     * <p>
+     * Another quirk is that this method is allowed to release resources "eventually",
+     * i.e., not before (in the happens before order) returning.
+     * Nevertheless, {@link #isClosed()} called after (in the happens-before order) {@link #close()} must return {@code true}.
+     *
+     * @see #close()
+     */
     @Override
     void close();
 }

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

@@ -24,14 +24,18 @@
 import com.mongodb.internal.binding.AsyncConnectionSource;
 import com.mongodb.internal.binding.AsyncReadBinding;
 import com.mongodb.internal.operation.OperationHelper.AsyncCallableWithSource;
+import com.mongodb.lang.NonNull;
 import org.bson.BsonDocument;
 import org.bson.BsonTimestamp;
 import org.bson.RawBsonDocument;
 
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
 
+import static com.mongodb.assertions.Assertions.assertNotNull;
+import static com.mongodb.assertions.Assertions.assertNull;
 import static com.mongodb.internal.async.ErrorHandlingResultCallback.errorHandlingCallback;
 import static com.mongodb.internal.operation.ChangeStreamBatchCursorHelper.isRetryableError;
 import static com.mongodb.internal.operation.OperationHelper.LOGGER;
@@ -44,7 +48,12 @@ final class AsyncChangeStreamBatchCursor<T> implements AsyncAggregateResponseBat
     private final int maxWireVersion;
 
     private volatile BsonDocument resumeToken;
-    private volatile AsyncAggregateResponseBatchCursor<RawBsonDocument> wrapped;
+    /**
+     * {@linkplain ChangeStreamBatchCursorHelper#isRetryableError(Throwable, int) Retryable errors} can result in
+     * {@code wrapped} containing {@code null} and {@link #isClosed} being {@code false}.
+     * This represents a situation in which the wrapped object was closed by {@code this} but {@code this} remained open.
+     */
+    private final AtomicReference<AsyncAggregateResponseBatchCursor<RawBsonDocument>> wrapped;
     private final AtomicBoolean isClosed;
 
     AsyncChangeStreamBatchCursor(final ChangeStreamOperation<T> changeStreamOperation,
@@ -53,16 +62,17 @@ final class AsyncChangeStreamBatchCursor<T> implements AsyncAggregateResponseBat
                                  final BsonDocument resumeToken,
                                  final int maxWireVersion) {
         this.changeStreamOperation = changeStreamOperation;
-        this.wrapped = wrapped;
+        this.wrapped = new AtomicReference<>(assertNotNull(wrapped));
         this.binding = binding;
         binding.retain();
         this.resumeToken = resumeToken;
         this.maxWireVersion = maxWireVersion;
         isClosed = new AtomicBoolean();
     }
 
+    @NonNull
     AsyncAggregateResponseBatchCursor<RawBsonDocument> getWrapped() {
-        return wrapped;
+        return assertNotNull(wrapped.get());
     }
 
     @Override
@@ -93,7 +103,7 @@ public void apply(final AsyncAggregateResponseBatchCursor<RawBsonDocument> curso
     public void close() {
         if (isClosed.compareAndSet(false, true)) {
             try {
-                wrapped.close();
+                nullifyAndCloseWrapped();
             } finally {
                 binding.release();
             }
@@ -102,22 +112,63 @@ public void close() {
 
     @Override
     public void setBatchSize(final int batchSize) {
-        wrapped.setBatchSize(batchSize);
+        getWrapped().setBatchSize(batchSize);
     }
 
     @Override
     public int getBatchSize() {
-        return wrapped.getBatchSize();
+        return getWrapped().getBatchSize();
     }
 
     @Override
     public boolean isClosed() {
-        return isClosed.get();
+        if (isClosed.get()) {
+            return true;
+        } else if (wrappedClosedItself()) {
+            close();
+            return true;
+        } else {
+            return false;
+        }
+    }
+
+    private boolean wrappedClosedItself() {
+        AsyncAggregateResponseBatchCursor<RawBsonDocument> observedWrapped = wrapped.get();
+        return observedWrapped != null && observedWrapped.isClosed();
+    }
+
+    /**
+     * {@code null} is written to {@link #wrapped} before closing the wrapped object to maintain the following guarantee:
+     * if {@link #wrappedClosedItself()} observes a {@linkplain AsyncAggregateResponseBatchCursor#isClosed() closed} wrapped object,
+     * then it closed itself as opposed to being closed by {@code this}.
+     */
+    private void nullifyAndCloseWrapped() {
+        AsyncAggregateResponseBatchCursor<RawBsonDocument> observedWrapped = wrapped.getAndSet(null);
+        if (observedWrapped != null) {
+            observedWrapped.close();
+        }
+    }
+
+    /**
+     * This method guarantees that the {@code newValue} argument is closed even if
+     * {@link #setWrappedOrCloseIt(AsyncAggregateResponseBatchCursor)} is called concurrently with or after (in the happens-before order)
+     * the method {@link #close()}.
+     */
+    private void setWrappedOrCloseIt(final AsyncAggregateResponseBatchCursor<RawBsonDocument> newValue) {
+        if (isClosed()) {
+            assertNull(this.wrapped.get());
+            newValue.close();
+        } else {
+            assertNull(this.wrapped.getAndSet(newValue));
+            if (isClosed()) {
+                nullifyAndCloseWrapped();
+            }
+        }
     }
 
     @Override
     public BsonDocument getPostBatchResumeToken() {
-        return wrapped.getPostBatchResumeToken();
+        return getWrapped().getPostBatchResumeToken();
     }
 
     @Override
@@ -127,7 +178,7 @@ public BsonTimestamp getOperationTime() {
 
     @Override
     public boolean isFirstBatchEmpty() {
-        return wrapped.isFirstBatchEmpty();
+        return getWrapped().isFirstBatchEmpty();
     }
 
     @Override
@@ -178,18 +229,18 @@ private interface AsyncBlock {
 
     private void resumeableOperation(final AsyncBlock asyncBlock, final SingleResultCallback<List<RawBsonDocument>> callback,
                                      final boolean tryNext) {
-        if (isClosed.get()) {
+        if (isClosed()) {
             callback.onResult(null, new MongoException(format("%s called after the cursor was closed.",
                     tryNext ? "tryNext()" : "next()")));
             return;
         }
-        asyncBlock.apply(wrapped, new SingleResultCallback<List<RawBsonDocument>>() {
+        asyncBlock.apply(getWrapped(), new SingleResultCallback<List<RawBsonDocument>>() {
             @Override
             public void onResult(final List<RawBsonDocument> result, final Throwable t) {
                 if (t == null) {
                     callback.onResult(result, null);
                 } else if (isRetryableError(t, maxWireVersion)) {
-                    wrapped.close();
+                    nullifyAndCloseWrapped();
                     retryOperation(asyncBlock, callback, tryNext);
                 } else {
                     callback.onResult(null, t);
@@ -214,9 +265,15 @@ public void onResult(final AsyncBatchCursor<T> result, final Throwable t) {
                             if (t != null) {
                                 callback.onResult(null, t);
                             } else {
-                                wrapped = ((AsyncChangeStreamBatchCursor<T>) result).getWrapped();
-                                binding.release(); // release the new change stream batch cursor's reference to the binding
-                                resumeableOperation(asyncBlock, callback, tryNext);
+                                try {
+                                    setWrappedOrCloseIt(((AsyncChangeStreamBatchCursor<T>) result).getWrapped());
+                                } finally {
+                                    try {
+                                        binding.release(); // release the new change stream batch cursor's reference to the binding
+                                    } finally {
+                                        resumeableOperation(asyncBlock, callback, tryNext);
+                                    }
+                                }
                             }
                         }
                     });

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

@@ -41,7 +41,7 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
-import static com.mongodb.assertions.Assertions.isTrue;
+import static com.mongodb.assertions.Assertions.assertFalse;
 import static com.mongodb.assertions.Assertions.isTrueArgument;
 import static com.mongodb.assertions.Assertions.notNull;
 import static com.mongodb.internal.async.ErrorHandlingResultCallback.errorHandlingCallback;
@@ -114,6 +114,14 @@ class AsyncQueryBatchCursor<T> implements AsyncAggregateResponseBatchCursor<T> {
         this.maxWireVersion = connection == null ? 0 : connection.getDescription().getMaxWireVersion();
     }
 
+    /**
+     * {@inheritDoc}
+     * <p>
+     * From the perspective of the code external to this class, this method is idempotent as required by its specification.
+     * However, if this method sets {@link #isClosePending},
+     * then it must be called by {@code this} again to release resources.
+     * This behavior does not violate externally observable idempotence because this method is allowed to release resources "eventually".
+     */
     @Override
     public void close() {
         boolean doClose = false;
@@ -145,24 +153,20 @@ public void tryNext(final SingleResultCallback<List<T>> callback) {
 
     @Override
     public void setBatchSize(final int batchSize) {
-        synchronized (this) {
-            isTrue("open", !isClosed);
-        }
+        assertFalse(isClosed());
         this.batchSize = batchSize;
     }
 
     @Override
     public int getBatchSize() {
-        synchronized (this) {
-            isTrue("open", !isClosed);
-        }
+        assertFalse(isClosed());
         return batchSize;
     }
 
     @Override
     public boolean isClosed() {
         synchronized (this) {
-            return isClosed;
+            return isClosed || isClosePending;
         }
     }
 
@@ -197,23 +201,18 @@ private void next(final SingleResultCallback<List<T>> callback, final boolean tr
                 results = null;
             }
             firstBatch = null;
-            ServerCursor localCursor = getServerCursor();
-            if (localCursor == null) {
-                synchronized (this) {
-                    isClosed = true;
-                }
+            if (getServerCursor() == null) {
+                close();
             }
             callback.onResult(results, null);
         } else {
             ServerCursor localCursor = getServerCursor();
             if (localCursor == null) {
-                synchronized (this) {
-                    isClosed = true;
-                }
+                close();
                 callback.onResult(null, null);
             } else {
                 synchronized (this) {
-                    if (isClosed) {
+                    if (isClosed()) {
                         callback.onResult(null, new MongoException(format("%s called after the cursor was closed.",
                                 tryNext ? "tryNext()" : "next()")));
                         return;