SERVER-34090 Allow resuming change stream when resume token's document key does not contain the shard key

1 files changed, 6 insertions, 35 deletions

diff --git a/jstests/noPassthrough/change_streams_require_majority_read_concern.js b/jstests/noPassthrough/change_streams_require_majority_read_concern.js
index 6ee9bd5c983..c1b142f9118 100644
--- a/jstests/noPassthrough/change_streams_require_majority_read_concern.js
+++ b/jstests/noPassthrough/change_streams_require_majority_read_concern.js
@@ -3,6 +3,7 @@
     "use strict";
 
     load("jstests/replsets/rslib.js");           // For startSetIfSupportsReadMajority.
+    load("jstests/libs/change_stream_util.js");  // For ChangeStreamTest.
     load("jstests/libs/write_concern_util.js");  // For stopReplicationOnSecondaries.
     const rst = new ReplSetTest({nodes: 2, nodeOptions: {enableMajorityReadConcern: ""}});
 
@@ -22,7 +23,6 @@
 
     rst.initiate();
 
-    const oplogProjection = {$project: {"_id.clusterTime": 0}};
     const name = "change_stream_require_majority_read_concern";
     const db = rst.getPrimary().getDB(name);
 
@@ -30,38 +30,8 @@
         return ns.split(/\.(.+)/)[1];
     }
 
-    // Helpers for testing that pipeline returns correct set of results.  Run startWatchingChanges
-    // with the pipeline, then insert the changes, then run assertNextBatchMatches with the result
-    // of startWatchingChanges and the expected set of results.
-    function startWatchingChanges({pipeline, collection, includeTs, aggregateOptions}) {
-        aggregateOptions = aggregateOptions || {cursor: {}};
-
-        if (!includeTs) {
-            // Strip the oplog fields we aren't testing.
-            pipeline.push(oplogProjection);
-        }
-
-        let res = assert.commandWorked(db.runCommand(
-            Object.merge({aggregate: collection.getName(), pipeline: pipeline}, aggregateOptions)));
-        assert.neq(res.cursor.id, 0);
-        return res.cursor;
-    }
-
-    // Gets one document from the cursor using getMore with awaitData disabled. Asserts if no
-    // document is present.
-    function getOneDoc(cursor) {
-        assert.commandWorked(db.adminCommand(
-            {configureFailPoint: "disableAwaitDataForGetMoreCmd", mode: "alwaysOn"}));
-        let res = assert.commandWorked(db.runCommand({
-            getMore: cursor.id,
-            collection: getCollectionNameFromFullNamespace(cursor.ns),
-            batchSize: 1
-        }));
-        assert.eq(res.cursor.nextBatch.length, 1);
-        assert.commandWorked(
-            db.adminCommand({configureFailPoint: "disableAwaitDataForGetMoreCmd", mode: "off"}));
-        return res.cursor.nextBatch[0];
-    }
+    // Use ChangeStreamTest to verify that the pipeline returns expected results.
+    const cst = new ChangeStreamTest(db);
 
     // Attempts to get a document from the cursor with awaitData disabled, and asserts if a
     // document is present.
@@ -108,7 +78,8 @@
     // Test not specifying readConcern defaults to "majority" read concern.
     stopReplicationOnSecondaries(rst);
     // Verify that the document just inserted cannot be returned.
-    let cursor = startWatchingChanges({pipeline: [{$changeStream: {}}], collection: primaryColl});
+    let cursor =
+        cst.startWatchingChanges({pipeline: [{$changeStream: {}}], collection: primaryColl});
     assert.eq(cursor.firstBatch.length, 0);
 
     // Insert a document on the primary only.
@@ -120,7 +91,7 @@
     rst.awaitLastOpCommitted();
 
     // Verify that the expected doc is returned because it has been committed.
-    let doc = getOneDoc(cursor);
+    let doc = cst.getOneChange(cursor);
     assert.docEq(doc.operationType, "insert");
     assert.docEq(doc.fullDocument, {_id: 2});
     rst.stopSet();