Restrict when startAtOperationTime is included for change streams

When resuming a change stream, the driver is required to save the
operationTime returned in the aggregate command response and send
it on subsequent aggregate commands when resuming the change stream,
until a resumeToken is available.

This commit ensures that it's only done when connected to a MongoDB 4.0
server, as per specification.

JAVA-2892

Author: jyemin

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

@@ -40,9 +40,7 @@ final class AsyncChangeStreamBatchCursor<T> implements AsyncBatchCursor<T> {
     AsyncChangeStreamBatchCursor(final ChangeStreamOperation<T> changeStreamOperation,
                                  final AsyncBatchCursor<RawBsonDocument> wrapped,
                                  final AsyncReadBinding binding) {
-        if (changeStreamOperation.getResumeToken() == null && changeStreamOperation.getStartAtOperationTime() == null) {
-            changeStreamOperation.startAtOperationTime(binding.getSessionContext().getOperationTime());
-        }
+        changeStreamOperation.setStartOperationTimeForResume(binding.getSessionContext().getOperationTime());
         this.changeStreamOperation = changeStreamOperation;
         this.resumeToken = changeStreamOperation.getResumeToken();
         this.wrapped = wrapped;
@@ -117,7 +115,8 @@ public void onResult(final List<RawBsonDocument> result, final Throwable t) {
 
     private void retryOperation(final AsyncBlock asyncBlock, final SingleResultCallback<List<RawBsonDocument>> callback) {
         if (resumeToken != null) {
-            changeStreamOperation.startAtOperationTime(null).resumeAfter(resumeToken);
+            changeStreamOperation.setStartOperationTimeForResume(null);
+            changeStreamOperation.resumeAfter(resumeToken);
         }
         changeStreamOperation.executeAsync(binding, new SingleResultCallback<AsyncBatchCursor<T>>() {
             @Override

driver-core/src/main/com/mongodb/operation/ChangeStreamBatchCursor.java

@@ -40,9 +40,7 @@ final class ChangeStreamBatchCursor<T> implements BatchCursor<T> {
     ChangeStreamBatchCursor(final ChangeStreamOperation<T> changeStreamOperation,
                             final BatchCursor<RawBsonDocument> wrapped,
                             final ReadBinding binding) {
-        if (changeStreamOperation.getResumeToken() == null && changeStreamOperation.getStartAtOperationTime() == null) {
-            changeStreamOperation.startAtOperationTime(binding.getSessionContext().getOperationTime());
-        }
+        changeStreamOperation.setStartOperationTimeForResume(binding.getSessionContext().getOperationTime());
         this.changeStreamOperation = changeStreamOperation;
         this.resumeToken = changeStreamOperation.getResumeToken();
         this.wrapped = wrapped;
@@ -141,7 +139,8 @@ <R> R resumeableOperation(final Function<BatchCursor<RawBsonDocument>, R> functi
             wrapped.close();
 
             if (resumeToken != null) {
-                changeStreamOperation.startAtOperationTime(null).resumeAfter(resumeToken);
+                changeStreamOperation.setStartOperationTimeForResume(null);
+                changeStreamOperation.resumeAfter(resumeToken);
             }
             wrapped = ((ChangeStreamBatchCursor<T>) changeStreamOperation.execute(binding)).getWrapped();
             binding.release(); // release the new change stream batch cursor's reference to the binding

driver-core/src/main/com/mongodb/operation/ChangeStreamOperation.java

@@ -42,6 +42,7 @@
 import java.util.concurrent.TimeUnit;
 
 import static com.mongodb.assertions.Assertions.notNull;
+import static com.mongodb.internal.operation.ServerVersionHelper.serverIsAtLeastVersionFourDotZero;
 
 /**
  * An operation that executes an {@code $changeStream} aggregation.
@@ -60,6 +61,7 @@ public class ChangeStreamOperation<T> implements AsyncReadOperation<AsyncBatchCu
 
     private BsonDocument resumeToken;
     private BsonTimestamp startAtOperationTime;
+    private boolean startAtOperationTimeForResume;
 
     /**
      * Construct a new instance.
@@ -240,6 +242,17 @@ public ChangeStreamOperation<T> startAtOperationTime(final BsonTimestamp startAt
         return this;
     }
 
+
+    void setStartOperationTimeForResume(final BsonTimestamp startAtOperationTime) {
+        if (startAtOperationTime == null && startAtOperationTimeForResume) {
+            this.startAtOperationTime = null;
+            startAtOperationTimeForResume = false;
+        } else if (this.startAtOperationTime == null && resumeToken == null)  {
+            this.startAtOperationTime = startAtOperationTime;
+            startAtOperationTimeForResume = true;
+        }
+    }
+
     /**
      * Returns the start at operation time
      *
@@ -296,7 +309,9 @@ public BsonArray create(final ConnectionDescription description, final SessionCo
                 }
 
                 if (startAtOperationTime != null) {
-                    changeStream.append("startAtOperationTime", startAtOperationTime);
+                    if (!startAtOperationTimeForResume || serverIsAtLeastVersionFourDotZero(description)) {
+                        changeStream.append("startAtOperationTime", startAtOperationTime);
+                    }
                 }
 
                 changeStreamPipeline.add(new BsonDocument("$changeStream", changeStream));
@@ -305,5 +320,4 @@ public BsonArray create(final ConnectionDescription description, final SessionCo
             }
         };
     }
-
 }

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

@@ -455,7 +455,7 @@ class ChangeStreamOperationSpecification extends OperationFunctionalSpecificatio
         def binding = Stub(ReadBinding) {
             getSessionContext() >> Stub(SessionContext) {
                 getReadConcern() >> ReadConcern.DEFAULT
-                getOperationTime() >> new BsonTimestamp()
+                getOperationTime() >> new BsonTimestamp(42, 1)
             }
             getReadConnectionSource() >> Stub(ConnectionSource) {
                 getConnection() >> Stub(Connection) {
@@ -479,7 +479,7 @@ class ChangeStreamOperationSpecification extends OperationFunctionalSpecificatio
         operation.execute(binding)
 
         then:
-        operation.getStartAtOperationTime() == new BsonTimestamp()
+        operation.getStartAtOperationTime() == new BsonTimestamp(42, 1)
 
         when: 'Set time'
         def startAtTime = new BsonTimestamp(42)