SERVER-44733 Change stream should throw ChangeStreamFatalError if a single shard cannot be targeted for updateLookup

(cherry picked from commit ccecd50087d22d90df71ab3c5dd4a58905590307)
(cherry picked from commit 20d508fbf59b05c6728162e85a9abfd06d41d0cd)

2 files changed, 68 insertions, 2 deletions

diff --git a/jstests/sharding/change_streams_update_lookup_shard_metadata_missing.js b/jstests/sharding/change_streams_update_lookup_shard_metadata_missing.js
new file mode 100644
index 00000000000..1018ba5fab2
--- /dev/null
+++ b/jstests/sharding/change_streams_update_lookup_shard_metadata_missing.js
@@ -0,0 +1,66 @@
+/**
+ * Tests that an updateLookup change stream throws ChangeStreamFatalError when it encounters an
+ * oplog entry whose documentKey omits the shard key.
+ * TODO SERVER-44598: the oplog entry will no longer omit the shard key when SERVER-44598 is fixed,
+ * and so this test will no longer be relevant.
+ * @tags: [uses_change_streams]
+ */
+(function() {
+    "use strict";
+
+    // The UUID consistency check can hit NotMasterNoSlaveOk when it attempts to obtain a list of
+    // collections from the shard Primaries through mongoS at the end of this test.
+    TestData.skipCheckingUUIDsConsistentAcrossCluster = true;
+
+    // Start a new sharded cluster with 2 nodes and obtain references to the test DB and collection.
+    const st = new ShardingTest({
+        shards: 2,
+        mongos: 1,
+        rs: {nodes: 3, setParameter: {writePeriodicNoops: true, periodicNoopIntervalSecs: 1}}
+    });
+
+    const mongosDB = st.s.getDB(jsTestName());
+    const mongosColl = mongosDB.test;
+    const shard0 = st.rs0;
+
+    // Enable sharding on the the test database and ensure that the primary is shard0.
+    assert.commandWorked(mongosDB.adminCommand({enableSharding: mongosDB.getName()}));
+    st.ensurePrimaryShard(mongosDB.getName(), shard0.getURL());
+
+    // Shard the source collection on {a: 1}, split across the shards at {a: 0}.
+    st.shardColl(mongosColl, {a: 1}, {a: 0}, {a: 1});
+
+    // Open a change stream on the collection.
+    const csCursor = mongosColl.watch();
+
+    // Write one document onto shard0 and obtain its resume token.
+    assert.writeOK(mongosColl.insert({_id: 0, a: -100}));
+    assert.soon(() => csCursor.hasNext());
+
+    const resumeToken = csCursor.next()._id;
+
+    // Step up one of the Secondaries, which will not have any sharding metadata loaded.
+    assert.commandWorked(shard0.getSecondary().adminCommand({replSetStepUp: 1}));
+    shard0.awaitNodesAgreeOnPrimary();
+
+    // Do a {multi:true} update. This will scatter to all shards and update the document on shard0.
+    // Because no metadata is loaded, this will write the update into the oplog with a documentKey
+    // containing only the _id field.
+    assert.soonNoExcept(
+        () => assert.writeOK(mongosColl.update({_id: 0}, {$set: {updated: true}}, false, true)));
+
+    // Resume the change stream with {fullDocument: 'updateLookup'}.
+    const cmdRes = assert.commandWorked(mongosColl.runCommand("aggregate", {
+        pipeline: [{$changeStream: {resumeAfter: resumeToken, fullDocument: "updateLookup"}}],
+        cursor: {}
+    }));
+
+    // Begin pulling from the stream. We should hit a ChangeStreamFatalError when the updateLookup
+    // attempts to read the update entry that is missing the shard key value of the document.
+    assert.soonNoExcept(
+        () => assert.commandFailedWithCode(
+            mongosColl.runCommand({getMore: cmdRes.cursor.id, collection: mongosColl.getName()}),
+            ErrorCodes.ChangeStreamFatalError));
+
+    st.stop();
+})();
\ No newline at end of file
diff --git a/jstests/sharding/resume_change_stream.js b/jstests/sharding/resume_change_stream.js
index 0e885550350..360b54a26dd 100644
--- a/jstests/sharding/resume_change_stream.js
+++ b/jstests/sharding/resume_change_stream.js
@@ -118,7 +118,7 @@
     ChangeStreamTest.assertChangeStreamThrowsCode({
         collection: mongosColl,
         pipeline: [{$changeStream: {resumeAfter: resumeTokenFromFirstUpdateOnShard0}}],
-        expectedCode: 40576
+        expectedCode: ErrorCodes.ChangeStreamFatalError
     });
 
     // Test that the change stream can't resume if the resume token *is* present in the oplog, but
@@ -129,7 +129,7 @@
     ChangeStreamTest.assertChangeStreamThrowsCode({
         collection: mongosColl,
         pipeline: [{$changeStream: {resumeAfter: resumeTokenFromFirstUpdateOnShard0}}],
-        expectedCode: 40576
+        expectedCode: ErrorCodes.ChangeStreamFatalError
     });
 
     // Drop the collection.