ChangeStream spec Resumable Error definition is too broad

JAVA-3293

Author: jstewart148

driver-async/src/test/functional/com/mongodb/async/client/ChangeStreamsTest.java

@@ -47,6 +47,7 @@
 import java.net.URISyntaxException;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Queue;
@@ -59,6 +60,7 @@
 import static java.util.Collections.singletonList;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
 import static org.junit.Assume.assumeFalse;
 import static org.junit.Assume.assumeNotNull;
 
@@ -143,8 +145,20 @@ public void shouldPassAllOutcomes() {
     private void checkStreamValues(final BsonDocument result, final AsyncBatchCursor<ChangeStreamDocument<BsonDocument>> cursor){
 
         BsonArray expectedResults = result.getArray("success", new BsonArray());
+        Queue<ChangeStreamDocument<BsonDocument>> results = null;
 
-        Queue<ChangeStreamDocument<BsonDocument>> results = getResults(cursor);
+        try {
+            results = getResults(cursor);
+        } catch (MongoException e) {
+            if (result.containsKey("error")) {
+                final BsonDocument error = result.getDocument("error");
+                assertTrue(e.getCode() == error.getInt32("code").intValue()
+                        || !Collections.disjoint(e.getErrorLabels(), error.getArray("errorLabels")));
+                return;
+            } else {
+                throw e;
+            }
+        }
         for (BsonValue expectedResult : expectedResults) {
             BsonDocument expected = expectedResult.asDocument();
 
@@ -195,7 +209,7 @@ private AsyncBatchCursor<ChangeStreamDocument<BsonDocument>> createCursor(final
     }
 
     private void checkExpectations() {
-        if (definition.getArray("expectations").size() > 0) {
+        if (definition.containsKey("expectations") && definition.getArray("expectations").size() > 0) {
 
             String database = definition.getString("target").getValue().equals("client") ? "admin" : namespace.getDatabaseName();
             List<CommandEvent> expectedEvents = getExpectedEvents(definition.getArray("expectations"), database, new BsonDocument());

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

@@ -22,12 +22,14 @@
 import com.mongodb.MongoNotPrimaryException;
 import com.mongodb.MongoSocketException;
 
+import java.util.Collections;
 import java.util.List;
 
 import static java.util.Arrays.asList;
 
 final class ChangeStreamBatchCursorHelper {
     private static final List<Integer> UNRETRYABLE_SERVER_ERROR_CODES = asList(136, 237, 280, 11601);
+    private static final List<String> NONRESUMABLE_CHANGE_STREAM_ERROR_LABELS = asList("NonResumableChangeStreamError");
 
     static boolean isRetryableError(final Throwable t) {
         if (!(t instanceof MongoException) || t instanceof MongoChangeStreamException) {
@@ -36,7 +38,8 @@ static boolean isRetryableError(final Throwable t) {
                 || t instanceof MongoSocketException) {
             return true;
         } else {
-            return !UNRETRYABLE_SERVER_ERROR_CODES.contains(((MongoException) t).getCode());
+            return !UNRETRYABLE_SERVER_ERROR_CODES.contains(((MongoException) t).getCode())
+                    && Collections.disjoint(NONRESUMABLE_CHANGE_STREAM_ERROR_LABELS, ((MongoException) t).getErrorLabels());
         }
     }
 

driver-core/src/test/resources/change-streams/change-streams-errors.json

@@ -54,9 +54,7 @@
               "cursor": {},
               "pipeline": [
                 {
-                  "$changeStream": {
-                    "fullDocument": "default"
-                  }
+                  "$changeStream": {}
                 },
                 {
                   "$unsupported": "foo"
@@ -73,6 +71,43 @@
           "code": 40324
         }
       }
+    },
+    {
+      "description": "Change Stream should error when _id is projected out",
+      "minServerVersion": "4.1.11",
+      "target": "collection",
+      "topology": [
+        "replicaset",
+        "sharded"
+      ],
+      "changeStreamPipeline": [
+        {
+          "$project": {
+            "_id": 0
+          }
+        }
+      ],
+      "changeStreamOptions": {},
+      "operations": [
+        {
+          "database": "change-stream-tests",
+          "collection": "test",
+          "name": "insertOne",
+          "arguments": {
+            "document": {
+              "z": 3
+            }
+          }
+        }
+      ],
+      "result": {
+        "error": {
+          "code": 280,
+          "errorLabels": [
+            "NonResumableChangeStreamError"
+          ]
+        }
+      }
     }
   ]
 }

driver-sync/src/test/functional/com/mongodb/client/ChangeStreamsTest.java

@@ -44,6 +44,7 @@
 import java.net.URISyntaxException;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.List;
 
 import static com.mongodb.JsonTestServerVersionChecker.skipTest;
@@ -52,6 +53,7 @@
 import static java.lang.String.format;
 import static java.util.Collections.singletonList;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
 import static org.junit.Assume.assumeFalse;
 
 // See https://github.com/mongodb/specifications/tree/master/source/change-streams/tests
@@ -154,6 +156,15 @@ private void checkStreamValues(final BsonDocument result, final MongoCursor<Chan
 
             assertEquals(expected.get("fullDocument"), actual.getFullDocument());
         }
+        if (result.containsKey("error")) {
+            BsonDocument error = result.getDocument("error");
+            try {
+                cursor.next();
+            } catch (MongoException e) {
+                assertTrue(e.getCode() == error.getInt32("code").intValue()
+                        || !Collections.disjoint(e.getErrorLabels(), error.getArray("errorLabels")));
+            }
+        }
     }
 
     @Nullable
@@ -169,7 +180,7 @@ private MongoCursor<ChangeStreamDocument<BsonDocument>> createCursor(final BsonD
     }
 
     private void checkExpectations() {
-        if (definition.getArray("expectations").size() > 0) {
+        if (definition.containsKey("expectations") && definition.getArray("expectations").size() > 0) {
 
             String database = definition.getString("target").getValue().equals("client") ? "admin" : namespace.getDatabaseName();
             List<CommandEvent> expectedEvents = getExpectedEvents(definition.getArray("expectations"), database, new BsonDocument());