Use white list for change stream resumability

JAVA-3639

update

Author: jstewart148

driver-core/src/main/com/mongodb/MongoBulkWriteException.java

@@ -77,14 +77,10 @@ public MongoBulkWriteException(final BulkWriteResult writeResult, final List<Bul
         this.writeConcernError = writeConcernError;
         this.serverAddress = serverAddress;
 
-        for (final String errorLabel : errorLabels) {
-            addLabel(errorLabel);
-        }
+        addLabels(errorLabels);
 
         if (writeConcernError != null) {
-            for (final String errorLabel : writeConcernError.getErrorLabels()) {
-                addLabel(errorLabel);
-            }
+            addLabels(writeConcernError.getErrorLabels());
         }
     }
 

driver-core/src/main/com/mongodb/MongoCommandException.java

@@ -20,7 +20,6 @@
 import org.bson.BsonDocument;
 import org.bson.BsonInt32;
 import org.bson.BsonString;
-import org.bson.BsonValue;
 import org.bson.codecs.BsonDocumentCodec;
 import org.bson.codecs.EncoderContext;
 import org.bson.json.JsonWriter;
@@ -50,9 +49,7 @@ public MongoCommandException(final BsonDocument response, final ServerAddress ad
               format("Command failed with error %s: '%s' on server %s. The full response is %s", extractErrorCodeAndName(response),
                      extractErrorMessage(response), address, getResponseAsJson(response)), address);
         this.response = response;
-        for (BsonValue curErrorLabel : response.getArray("errorLabels", new BsonArray())) {
-            addLabel(curErrorLabel.asString().getValue());
-        }
+        addLabels(response.getArray("errorLabels", new BsonArray()));
     }
 
     /**

driver-core/src/main/com/mongodb/MongoException.java

@@ -17,7 +17,11 @@
 package com.mongodb;
 
 import com.mongodb.lang.Nullable;
+import org.bson.BsonArray;
+import org.bson.BsonDocument;
+import org.bson.BsonValue;
 
+import java.util.Collection;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.Set;
@@ -115,12 +119,22 @@ public MongoException(final int code, final String msg, final Throwable t) {
         super(msg, t);
         this.code = code;
         if (t instanceof MongoException) {
-            for (final String errorLabel : ((MongoException) t).getErrorLabels()) {
-                addLabel(errorLabel);
-            }
+            addLabels(((MongoException) t).getErrorLabels());
         }
     }
 
+    /**
+     * @param code     the error code
+     * @param msg      the message
+     * @param response the response
+     * @since 4.1
+     */
+    public MongoException(final int code, final String msg, final BsonDocument response) {
+        super(msg);
+        this.code = code;
+        addLabels(response.getArray("errorLabels", new BsonArray()));
+    }
+
     /**
      * Gets the exception code
      *
@@ -176,4 +190,15 @@ public boolean hasErrorLabel(final String errorLabel) {
         return errorLabels.contains(errorLabel);
     }
 
+    protected void addLabels(final BsonArray labels) {
+        for (final BsonValue errorLabel : labels) {
+            addLabel(errorLabel.asString().getValue());
+        }
+    }
+
+    protected void addLabels(final Collection<String> labels) {
+        for (final String errorLabel : labels) {
+            addLabel(errorLabel);
+        }
+    }
 }

driver-core/src/main/com/mongodb/MongoExecutionTimeoutException.java

@@ -16,6 +16,8 @@
 
 package com.mongodb;
 
+import org.bson.BsonDocument;
+
 /**
  * Exception indicating that the execution of the current operation timed out as a result of the maximum operation time being exceeded.
  *
@@ -34,4 +36,17 @@ public MongoExecutionTimeoutException(final int code, final String message) {
         super(code, message);
 
     }
+
+    /**
+     * Construct a new instance.
+     *
+     * @param code the error code
+     * @param message the error message
+     * @param response the response
+     * @since 4.1
+     */
+    public MongoExecutionTimeoutException(final int code, final String message, final BsonDocument response) {
+        super(code, message, response);
+
+    }
 }

driver-core/src/main/com/mongodb/MongoQueryException.java

@@ -48,9 +48,7 @@ public MongoQueryException(final ServerAddress address, final int errorCode, fin
      */
     public MongoQueryException(final MongoCommandException commandException) {
         this(commandException.getServerAddress(), commandException.getErrorCode(), commandException.getErrorMessage());
-        for (String label : commandException.getErrorLabels()) {
-            addLabel(label);
-        }
+        addLabels(commandException.getErrorLabels());
     }
 
     /**

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

@@ -46,4 +46,11 @@ public interface AsyncAggregateResponseBatchCursor<T> extends AsyncBatchCursor<T
      * @return true if the first batch was empty
      */
     boolean isFirstBatchEmpty();
+
+    /**
+     * Returns the max wire version.
+     *
+     * @return the max wire version
+     */
+    int getMaxWireVersion();
 }

driver-core/src/main/com/mongodb/internal/connection/ProtocolHelper.java

@@ -236,7 +236,7 @@ public static MongoException createSpecialException(final BsonDocument response,
         int errorCode = getErrorCode(response);
         String errorMessage = getErrorMessage(response, errorMessageFieldName);
         if (ErrorCategory.fromErrorCode(errorCode) == ErrorCategory.EXECUTION_TIMEOUT) {
-            return new MongoExecutionTimeoutException(errorCode, errorMessage);
+            return new MongoExecutionTimeoutException(errorCode, errorMessage, response);
         } else if (errorMessage.contains("not master or secondary") || errorMessage.contains("node is recovering")
                 || RECOVERING_CODES.contains(errorCode)) {
             return new MongoNodeIsRecoveringException(response, serverAddress);

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

@@ -49,4 +49,11 @@ public interface AggregateResponseBatchCursor<T> extends BatchCursor<T> {
      * @return true if the first batch was empty
      */
     boolean isFirstBatchEmpty();
+
+    /**
+     * Returns the max wire version.
+     *
+     * @return the max wire version
+     */
+    int getMaxWireVersion();
 }

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

@@ -40,6 +40,7 @@
 final class AsyncChangeStreamBatchCursor<T> implements AsyncAggregateResponseBatchCursor<T> {
     private final AsyncReadBinding binding;
     private final ChangeStreamOperation<T> changeStreamOperation;
+    private final int maxWireVersion;
 
     private volatile BsonDocument resumeToken;
     private volatile AsyncAggregateResponseBatchCursor<RawBsonDocument> wrapped;
@@ -53,12 +54,14 @@ final class AsyncChangeStreamBatchCursor<T> implements AsyncAggregateResponseBat
     AsyncChangeStreamBatchCursor(final ChangeStreamOperation<T> changeStreamOperation,
                                  final AsyncAggregateResponseBatchCursor<RawBsonDocument> wrapped,
                                  final AsyncReadBinding binding,
-                                 final BsonDocument resumeToken) {
+                                 final BsonDocument resumeToken,
+                                 final int maxWireVersion) {
         this.changeStreamOperation = changeStreamOperation;
         this.wrapped = wrapped;
         this.binding = binding;
         binding.retain();
         this.resumeToken = resumeToken;
+        this.maxWireVersion = maxWireVersion;
     }
 
     AsyncAggregateResponseBatchCursor<RawBsonDocument> getWrapped() {
@@ -141,6 +144,11 @@ public boolean isFirstBatchEmpty() {
         return wrapped.isFirstBatchEmpty();
     }
 
+    @Override
+    public int getMaxWireVersion() {
+        return maxWireVersion;
+    }
+
     private void cachePostBatchResumeToken(final AsyncAggregateResponseBatchCursor<RawBsonDocument> queryBatchCursor) {
         if (queryBatchCursor.getPostBatchResumeToken() != null) {
             resumeToken = queryBatchCursor.getPostBatchResumeToken();
@@ -209,7 +217,7 @@ public void onResult(final List<RawBsonDocument> result, final Throwable t) {
                 if (t == null) {
                     endOperationInProgress();
                     callback.onResult(result, null);
-                } else if (isRetryableError(t)) {
+                } else if (isRetryableError(t, maxWireVersion)) {
                     wrapped.close();
                     retryOperation(asyncBlock, callback, tryNext);
                 } else {

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

@@ -71,6 +71,7 @@ class AsyncQueryBatchCursor<T> implements AsyncAggregateResponseBatchCursor<T> {
     private volatile BsonDocument postBatchResumeToken;
     private volatile BsonTimestamp operationTime;
     private volatile boolean firstBatchEmpty;
+    private volatile int maxWireVersion = 0;
 
     /* protected by `this` */
     private boolean isOperationInProgress = false;
@@ -108,6 +109,9 @@ class AsyncQueryBatchCursor<T> implements AsyncAggregateResponseBatchCursor<T> {
                 killCursor(connection);
             }
         }
+        if (connection != null) {
+            this.maxWireVersion = connection.getDescription().getMaxWireVersion();
+        }
     }
 
     @Override
@@ -177,6 +181,11 @@ public boolean isFirstBatchEmpty() {
         return firstBatchEmpty;
     }
 
+    @Override
+    public int getMaxWireVersion() {
+        return maxWireVersion;
+    }
+
     private void next(final SingleResultCallback<List<T>> callback, final boolean tryNext) {
         if (isClosed()) {
             callback.onResult(null, new MongoException(format("%s called after the cursor was closed.",