SERVER-35029: Add change stream notification for database drop

13 files changed, 213 insertions, 65 deletions

diff --git a/jstests/change_streams/metadata_notifications.js b/jstests/change_streams/metadata_notifications.js
index ea7559de77b..ef17b2b4249 100644
--- a/jstests/change_streams/metadata_notifications.js
+++ b/jstests/change_streams/metadata_notifications.js
@@ -1,4 +1,7 @@
 // Tests of $changeStream notifications for metadata operations.
+// Do not run in whole-cluster passthrough since this test assumes that the change stream will be
+// invalidated by a database drop.
+// @tags: [do_not_run_in_whole_cluster_passthrough]
 (function() {
     "use strict";
 
@@ -27,6 +30,13 @@
     assert.neq(change.id, 0);
     assert.eq(change.nextBatch.length, 0, tojson(change.nextBatch));
 
+    // Dropping the empty database should not generate any notification for the change stream, since
+    // the collection does not exist yet.
+    assert.commandWorked(db.dropDatabase());
+    change = cst.getNextBatch(cursor);
+    assert.neq(change.id, 0);
+    assert.eq(change.nextBatch.length, 0, tojson(change.nextBatch));
+
     // After collection creation, we expect to see oplog entries for each subsequent operation.
     let coll = assertCreateCollection(db, collName);
     assert.writeOK(coll.insert({_id: 0}));
diff --git a/jstests/change_streams/whole_cluster.js b/jstests/change_streams/whole_cluster.js
index 9d154220b6e..894e90fc31d 100644
--- a/jstests/change_streams/whole_cluster.js
+++ b/jstests/change_streams/whole_cluster.js
@@ -119,14 +119,12 @@
         assertDropCollection(db.getSiblingDB(dbName), "test");
     });
 
-    // Dropping a database should generate drop entries for each collection followed by an
-    // invalidate.
-    // TODO SERVER-35029: This test should not invalidate the stream once there's support for
-    // returning a notification for the dropDatabase command.
+    // Dropping a database should generate drop entries for each collection followed by a database
+    // drop.
     assert.commandWorked(otherDB.dropDatabase());
     expected = [
         {operationType: "drop", ns: {db: otherDB.getName(), coll: "t2"}},
-        {operationType: "invalidate"}
+        {operationType: "dropDatabase", ns: {db: otherDB.getName()}},
     ];
 
     cst.assertNextChangesEqual({cursor: cursor, expectedChanges: expected});
diff --git a/jstests/change_streams/whole_cluster_metadata_notifications.js b/jstests/change_streams/whole_cluster_metadata_notifications.js
index 7c8e9c59f8f..471f31282ff 100644
--- a/jstests/change_streams/whole_cluster_metadata_notifications.js
+++ b/jstests/change_streams/whole_cluster_metadata_notifications.js
@@ -31,7 +31,8 @@
         assert.writeOK(coll.remove({_id: 1}));
     }
 
-    // Drop the second database, which should invalidate the stream.
+    // Drop the second database, which should generate a 'drop' entry for the collection followed
+    // by a 'dropDatabase' entry.
     assert.commandWorked(testDB2.dropDatabase());
 
     // We should get 6 oplog entries; three ops of type insert, update, delete from each database.
@@ -50,10 +51,8 @@
         cursor: aggCursor,
         expectedChanges: [
             {operationType: "drop", ns: {db: testDB2.getName(), coll: db2Coll.getName()}},
-            // TODO SERVER-35029: Return an entry for a database drop, instead of "invalidate".
-            {operationType: "invalidate"}
+            {operationType: "dropDatabase", ns: {db: testDB2.getName()}},
         ],
-        expectInvalidate: true
     });
 
     // Test that a cluster-wide change stream can be resumed using a token from a collection which
diff --git a/jstests/change_streams/whole_cluster_resumability.js b/jstests/change_streams/whole_cluster_resumability.js
index 9058c58010a..ad9145ba74a 100644
--- a/jstests/change_streams/whole_cluster_resumability.js
+++ b/jstests/change_streams/whole_cluster_resumability.js
@@ -6,8 +6,8 @@
     load("jstests/libs/change_stream_util.js");        // For ChangeStreamTest.
 
     // Create two databases, with one collection in each.
-    const testDBs = [db, db.getSiblingDB(jsTestName() + "_other")];
-    const[db1Coll, db2Coll] = testDBs.map((db) => assertDropAndRecreateCollection(db, "test"));
+    const testDBs = [db.getSiblingDB(jsTestName()), db.getSiblingDB(jsTestName() + "_other")];
+    let [db1Coll, db2Coll] = testDBs.map((db) => assertDropAndRecreateCollection(db, "test"));
     const adminDB = db.getSiblingDB("admin");
 
     let cst = new ChangeStreamTest(adminDB);
@@ -55,5 +55,47 @@
     });
     assert.docEq(cst.getOneChange(resumeCursor), thirdInsertChangeDoc);
 
+    // Dropping a database should generate a 'drop' notification for the collection followed by a
+    // 'dropDatabase' notification.
+    assert.commandWorked(testDBs[0].dropDatabase());
+    const dropDbChanges = cst.assertNextChangesEqual({
+        cursor: resumeCursor,
+        expectedChanges: [
+            {operationType: "drop", ns: {db: testDBs[0].getName(), coll: db1Coll.getName()}},
+            {operationType: "dropDatabase", ns: {db: testDBs[0].getName()}}
+        ]
+    });
+
+    // Recreate the database and verify that the change stream picks up another insert.
+    assert.writeOK(db1Coll.insert({_id: 5}));
+
+    let change = cst.getOneChange(resumeCursor);
+    assert.eq(change.operationType, "insert", tojson(change));
+    assert.eq(change.ns, {db: testDBs[0].getName(), coll: db1Coll.getName()}, tojson(change));
+    assert.eq(change.fullDocument, {_id: 5}, tojson(change));
+
+    // Resume the change stream from the 'drop' entry.
+    resumeCursor = cst.startWatchingChanges({
+        pipeline:
+            [{$changeStream: {resumeAfter: dropDbChanges[0]._id, allChangesForCluster: true}}],
+        collection: 1,
+        aggregateOptions: {cursor: {batchSize: 0}},
+    });
+    change = cst.getOneChange(resumeCursor);
+    assert.eq(change.operationType, "dropDatabase", tojson(change));
+    assert.eq(change.ns, {db: testDBs[0].getName()}, tojson(change));
+
+    // Resume the change stream from the 'dropDatabase' entry.
+    resumeCursor = cst.startWatchingChanges({
+        pipeline:
+            [{$changeStream: {resumeAfter: dropDbChanges[1]._id, allChangesForCluster: true}}],
+        collection: 1,
+        aggregateOptions: {cursor: {batchSize: 0}},
+    });
+    change = cst.getOneChange(resumeCursor);
+    assert.eq(change.operationType, "insert", tojson(change));
+    assert.eq(change.ns, {db: testDBs[0].getName(), coll: db1Coll.getName()}, tojson(change));
+    assert.eq(change.fullDocument, {_id: 5}, tojson(change));
+
     cst.cleanUp();
 })();
diff --git a/jstests/change_streams/whole_db.js b/jstests/change_streams/whole_db.js
index 8c287e8493f..82240dacd4b 100644
--- a/jstests/change_streams/whole_db.js
+++ b/jstests/change_streams/whole_db.js
@@ -1,4 +1,7 @@
 // Basic tests for $changeStream against all collections in a database.
+// Do not run in whole-cluster passthrough since this test assumes that the change stream will be
+// invalidated by a database drop.
+// @tags: [do_not_run_in_whole_cluster_passthrough]
 (function() {
     "use strict";
 
@@ -8,6 +11,7 @@
     load("jstests/libs/fixture_helpers.js");           // For FixtureHelpers.
 
     db = db.getSiblingDB(jsTestName());
+    assert.commandWorked(db.dropDatabase());
 
     // Test that a single-database change stream cannot be opened on "admin", "config", or "local".
     assertInvalidChangeStreamNss("admin", 1);
@@ -63,19 +67,5 @@
         cst.assertNextChangesEqual({cursor: cursor, expectedChanges: expected});
     });
 
-    // Dropping the database should generate collection drop entries followed by an invalidate. Note
-    // that the order of collection drops is not guaranteed so only check the database name.
-    assert.commandWorked(db.dropDatabase());
-    let change = cst.getOneChange(cursor);
-    assert.eq(change.operationType, "drop", tojson(change));
-    assert.eq(change.ns.db, db.getName(), tojson(change));
-    change = cst.getOneChange(cursor);
-    assert.eq(change.operationType, "drop", tojson(change));
-    assert.eq(change.ns.db, db.getName(), tojson(change));
-
-    // TODO SERVER-35029: Expect to see a 'dropDatabase' entry before the invalidate.
-    change = cst.getOneChange(cursor, true);
-    assert.eq(change.operationType, "invalidate", tojson(change));
-
     cst.cleanUp();
 }());
diff --git a/jstests/change_streams/whole_db_metadata_notifications.js b/jstests/change_streams/whole_db_metadata_notifications.js
index 6fa694aa1e8..0cb15d57680 100644
--- a/jstests/change_streams/whole_db_metadata_notifications.js
+++ b/jstests/change_streams/whole_db_metadata_notifications.js
@@ -1,4 +1,7 @@
 // Tests of metadata notifications for a $changeStream on a whole database.
+// Do not run in whole-cluster passthrough since this test assumes that the change stream will be
+// invalidated by a database drop.
+// @tags: [do_not_run_in_whole_cluster_passthrough]
 (function() {
     "use strict";
 
@@ -159,6 +162,15 @@
         assert.eq(0, cst.getNextBatch(aggCursor).nextBatch.length);
         assert.writeOK(coll.insert({_id: 2}));
         assert.eq(cst.getOneChange(aggCursor).operationType, "insert");
+
+        // Drop the new collection to avoid an additional 'drop' notification when the database is
+        // dropped.
+        assertDropCollection(testDB, "renamed_coll");
+        cst.assertNextChangesEqual({
+            cursor: aggCursor,
+            expectedChanges:
+                [{operationType: "drop", ns: {db: testDB.getName(), coll: "renamed_coll"}}],
+        });
     }
 
     // Dropping a collection should return a 'drop' entry.
@@ -228,6 +240,23 @@
     // Drop the "system.views" collection to avoid view catalog errors in subsequent tests.
     assertDropCollection(testDB, "system.views");
     assertDropCollection(testDB, "non_system_collection");
+    cst.assertNextChangesEqual({
+        cursor: aggCursor,
+        expectedChanges: [
+            {operationType: "drop", ns: {db: testDB.getName(), coll: "non_system_collection"}},
+        ]
+    });
+
+    // Dropping the database should generate a 'dropDatabase' notification followed by an
+    // 'invalidate'.
+    assert.commandWorked(testDB.dropDatabase());
+    cst.assertNextChangesEqual({
+        cursor: aggCursor,
+        expectedChanges: [
+            {operationType: "dropDatabase", ns: {db: testDB.getName()}},
+            {operationType: "invalidate"}
+        ]
+    });
 
     cst.cleanUp();
 }());
diff --git a/jstests/change_streams/whole_db_resumability.js b/jstests/change_streams/whole_db_resumability.js
index 682d0546cb7..25f03943112 100644
--- a/jstests/change_streams/whole_db_resumability.js
+++ b/jstests/change_streams/whole_db_resumability.js
@@ -1,4 +1,7 @@
 // Basic tests for resuming a $changeStream that is open against all collections in a database.
+// Do not run in whole-cluster passthrough since this test assumes that the change stream will be
+// invalidated by a database drop.
+// @tags: [do_not_run_in_whole_cluster_passthrough]
 (function() {
     "use strict";
 
@@ -6,10 +9,11 @@
     load("jstests/libs/change_stream_util.js");        // For ChangeStreamTest.
 
     // Drop and recreate the collections to be used in this set of tests.
-    const coll = assertDropAndRecreateCollection(db, jsTestName() + "resume_coll");
-    const otherColl = assertDropAndRecreateCollection(db, jsTestName() + "resume_coll_other");
+    const testDB = db.getSiblingDB(jsTestName());
+    let coll = assertDropAndRecreateCollection(testDB, "resume_coll");
+    const otherColl = assertDropAndRecreateCollection(testDB, "resume_coll_other");
 
-    let cst = new ChangeStreamTest(db);
+    let cst = new ChangeStreamTest(testDB);
     let resumeCursor = cst.startWatchingChanges({pipeline: [{$changeStream: {}}], collection: 1});
 
     // Insert a single document to each collection and save the resume token from the first insert.
@@ -17,7 +21,7 @@
     assert.writeOK(otherColl.insert({_id: 2}));
     const firstInsertChangeDoc = cst.getOneChange(resumeCursor);
     assert.docEq(firstInsertChangeDoc.fullDocument, {_id: 1});
-    assert.eq(firstInsertChangeDoc.ns, {db: "test", coll: coll.getName()});
+    assert.eq(firstInsertChangeDoc.ns, {db: testDB.getName(), coll: coll.getName()});
 
     // Test resuming the change stream after the first insert should pick up the insert on the
     // second collection.
@@ -29,13 +33,13 @@
 
     const secondInsertChangeDoc = cst.getOneChange(resumeCursor);
     assert.docEq(secondInsertChangeDoc.fullDocument, {_id: 2});
-    assert.eq(secondInsertChangeDoc.ns, {db: "test", coll: otherColl.getName()});
+    assert.eq(secondInsertChangeDoc.ns, {db: testDB.getName(), coll: otherColl.getName()});
 
     // Insert a third document to the first collection and test that the change stream picks it up.
     assert.writeOK(coll.insert({_id: 3}));
     const thirdInsertChangeDoc = cst.getOneChange(resumeCursor);
     assert.docEq(thirdInsertChangeDoc.fullDocument, {_id: 3});
-    assert.eq(thirdInsertChangeDoc.ns, {db: "test", coll: coll.getName()});
+    assert.eq(thirdInsertChangeDoc.ns, {db: testDB.getName(), coll: coll.getName()});
 
     // Test resuming after the first insert again.
     resumeCursor = cst.startWatchingChanges({
@@ -54,5 +58,69 @@
     });
     assert.docEq(cst.getOneChange(resumeCursor), thirdInsertChangeDoc);
 
+    // Explicitly drop one collection to ensure reliability of the order of notifications from the
+    // dropDatabase command.
+    assertDropCollection(testDB, otherColl.getName());
+    const firstCollDrop = cst.getOneChange(resumeCursor);
+    assert.eq(firstCollDrop.operationType, "drop", tojson(firstCollDrop));
+    assert.eq(firstCollDrop.ns, {db: testDB.getName(), coll: otherColl.getName()});
+
+    // Dropping a database should generate a 'drop' notification for each collection, a
+    // 'dropDatabase' notification, and finally an 'invalidate'.
+    assert.commandWorked(testDB.dropDatabase());
+    const expectedChangesAfterFirstDrop = [
+        {operationType: "drop", ns: {db: testDB.getName(), coll: coll.getName()}},
+        {operationType: "dropDatabase", ns: {db: testDB.getName()}},
+        {operationType: "invalidate"}
+    ];
+    const dropDbChanges = cst.assertNextChangesEqual(
+        {cursor: resumeCursor, expectedChanges: expectedChangesAfterFirstDrop});
+
+    // Resume from the first collection drop.
+    resumeCursor = cst.startWatchingChanges({
+        pipeline: [{$changeStream: {resumeAfter: firstCollDrop._id}}],
+        collection: 1,
+    });
+    cst.assertNextChangesEqual(
+        {cursor: resumeCursor, expectedChanges: expectedChangesAfterFirstDrop});
+
+    // Resume from the second collection drop.
+    resumeCursor = cst.startWatchingChanges({
+        pipeline: [{$changeStream: {resumeAfter: dropDbChanges[0]._id}}],
+        collection: 1,
+    });
+    cst.assertNextChangesEqual(
+        {cursor: resumeCursor, expectedChanges: expectedChangesAfterFirstDrop.slice(1)});
+
+    // Recreate the test collection.
+    coll = assertCreateCollection(testDB, coll.getName());
+    assert.writeOK(coll.insert({_id: 0}));
+
+    // Test resuming from the 'dropDatabase' entry.
+    // TODO SERVER-34789: Resuming from the 'dropDatabase' should return a single invalidate
+    // notification.
+    resumeCursor = cst.startWatchingChanges({
+        pipeline: [{$changeStream: {resumeAfter: dropDbChanges[1]._id}}],
+        collection: 1,
+        aggregateOptions: {cursor: {batchSize: 0}},
+    });
+    let change = cst.getOneChange(resumeCursor);
+    assert.eq(change.operationType, "insert", tojson(change));
+    assert.eq(change.fullDocument, {_id: 0}, tojson(change));
+    assert.eq(change.ns, {db: testDB.getName(), coll: coll.getName()}, tojson(change));
+
+    // Test resuming from the 'invalidate' entry.
+    // TODO SERVER-34789: Resuming from an invalidate should error or return an invalidate
+    // notification.
+    resumeCursor = cst.startWatchingChanges({
+        pipeline: [{$changeStream: {resumeAfter: dropDbChanges[2]._id}}],
+        collection: 1,
+        aggregateOptions: {cursor: {batchSize: 0}},
+    });
+    change = cst.getOneChange(resumeCursor);
+    assert.eq(change.operationType, "insert", tojson(change));
+    assert.eq(change.fullDocument, {_id: 0}, tojson(change));
+    assert.eq(change.ns, {db: testDB.getName(), coll: coll.getName()}, tojson(change));
+
     cst.cleanUp();
 })();
diff --git a/jstests/libs/change_stream_util.js b/jstests/libs/change_stream_util.js
index 48f1e133b53..71a56f3ffe8 100644
--- a/jstests/libs/change_stream_util.js
+++ b/jstests/libs/change_stream_util.js
@@ -3,13 +3,35 @@
  * opened cursors and kills them on cleanup.
  */
 
-// Helper function used internally by ChangeStreamTest. If no passthrough is active, it is exactly
-// the same as calling db.runCommand. If a passthrough is active and has defined a function
-// 'changeStreamPassthroughAwareRunCommand', then this method will be overridden to allow individual
-// streams to explicitly exempt themselves from being modified by the passthrough.
+/**
+ * Enumeration of the possible types of change streams.
+ */
+const ChangeStreamWatchMode = Object.freeze({
+    kCollection: 1,
+    kDb: 2,
+    kCluster: 3,
+});
+
+/**
+ * Helper function used internally by ChangeStreamTest. If no passthrough is active, it is exactly
+ * the same as calling db.runCommand. If a passthrough is active and has defined a function
+ * 'changeStreamPassthroughAwareRunCommand', then this method will be overridden to allow individual
+ * streams to explicitly exempt themselves from being modified by the passthrough.
+ */
 function isChangeStreamPassthrough() {
     return typeof changeStreamPassthroughAwareRunCommand != 'undefined';
 }
+
+/**
+ * Helper function to retrieve the type of change stream based on the passthrough suite in which the
+ * test is being run. If no passthrough is active, this method returns the kCollection watch mode.
+ */
+function changeStreamPassthroughType() {
+    return typeof ChangeStreamPassthroughHelpers === 'undefined'
+        ? ChangeStreamWatchMode.kCollection
+        : ChangeStreamPassthroughHelpers.passthroughType();
+}
+
 const runCommandChangeStreamPassthroughAware =
     (!isChangeStreamPassthrough() ? ((db, cmdObj) => db.runCommand(cmdObj))
                                   : changeStreamPassthroughAwareRunCommand);
@@ -22,19 +44,12 @@ const runCommandChangeStreamPassthroughAware =
  *     - whole cluster streams: none.
  */
 function assertInvalidateOp({cursor, opType}) {
-    if (!isChangeStreamPassthrough()) {
-        // All metadata operations will invalidate a single-collection change stream.
+    if (!isChangeStreamPassthrough() ||
+        (changeStreamPassthroughType() == ChangeStreamWatchMode.kDb && opType == "dropDatabase")) {
         assert.soon(() => cursor.hasNext());
         assert.eq(cursor.next().operationType, "invalidate");
         assert(cursor.isExhausted());
         assert(cursor.isClosed());
-    } else {
-        // Collection drops do not validate whole-db/cluster change streams.
-        if (opType == "drop") {
-            return;
-        }
-
-        // TODO SERVER-35029: Database drops should only invalidate whole-db change streams.
     }
 }
 
@@ -256,7 +271,7 @@ function ChangeStreamTest(_db, name = "ChangeStreamTest") {
 
     /**
      * Returns the document to be used for the value of a $changeStream stage, given a watchMode
-     * of type ChangeStreamTest.WatchMode and optional resumeAfter value.
+     * of type ChangeStreamWatchMode and optional resumeAfter value.
      */
     self.getChangeStreamStage = function(watchMode, resumeAfter) {
         const changeStreamDoc = {};
@@ -264,20 +279,20 @@ function ChangeStreamTest(_db, name = "ChangeStreamTest") {
             changeStreamDoc.resumeAfter = resumeAfter;
         }
 
-        if (watchMode == ChangeStreamTest.WatchMode.kCluster) {
+        if (watchMode == ChangeStreamWatchMode.kCluster) {
             changeStreamDoc.allChangesForCluster = true;
         }
         return changeStreamDoc;
     };
 
     /**
-     * Create a change stream of the given watch mode (see ChangeStreamTest.WatchMode) on the given
+     * Create a change stream of the given watch mode (see ChangeStreamWatchMode) on the given
      * collection. Will resume from a given point if resumeAfter is specified.
      */
     self.getChangeStream = function({watchMode, coll, resumeAfter}) {
         return self.startWatchingChanges({
             pipeline: [{$changeStream: self.getChangeStreamStage(watchMode, resumeAfter)}],
-            collection: (watchMode == ChangeStreamTest.WatchMode.kCollection ? coll : 1),
+            collection: (watchMode == ChangeStreamWatchMode.kCollection ? coll : 1),
             // Use a batch size of 0 to prevent any notifications from being returned in the first
             // batch. These would be ignored by ChangeStreamTest.getOneChange().
             aggregateOptions: {cursor: {batchSize: 0}},
@@ -319,23 +334,13 @@ ChangeStreamTest.assertChangeStreamThrowsCode = function assertChangeStreamThrow
  * the watchMode.
  */
 ChangeStreamTest.getDBForChangeStream = function(watchMode, dbObj) {
-    if (watchMode == ChangeStreamTest.WatchMode.kCluster) {
+    if (watchMode == ChangeStreamWatchMode.kCluster) {
         return dbObj.getSiblingDB("admin");
     }
     return dbObj;
 };
 
 /**
- * Used in getChangeStream() and getChangeStreamStage() helpers, for specifying which type of
- * changeStream to open.
- */
-ChangeStreamTest.WatchMode = Object.freeze({
-    kCollection: 1,
-    kDb: 2,
-    kCluster: 3,
-});
-
-/**
  * A set of functions to help validate the behaviour of $changeStreams for a given namespace.
  */
 function assertChangeStreamNssBehaviour(dbName, collName = "test", options, assertFunc) {
diff --git a/jstests/libs/override_methods/implicit_whole_cluster_changestreams.js b/jstests/libs/override_methods/implicit_whole_cluster_changestreams.js
index f6a1da78b4e..c7ba66763a2 100644
--- a/jstests/libs/override_methods/implicit_whole_cluster_changestreams.js
+++ b/jstests/libs/override_methods/implicit_whole_cluster_changestreams.js
@@ -47,6 +47,10 @@ ChangeStreamPassthroughHelpers.changeStreamSpec = function() {
     return {allChangesForCluster: true};
 };
 
+ChangeStreamPassthroughHelpers.passthroughType = function() {
+    return ChangeStreamWatchMode.kCluster;
+};
+
 // Redirect the DB's 'watch' function to use the cluster-wide version. The Collection.watch helper
 // has already been overridden to use DB.watch when we loaded 'implicit_whole_db_changestreams.js',
 // so this ensures that both the Collection and DB helpers will actually call the Mongo function.
diff --git a/jstests/libs/override_methods/implicit_whole_db_changestreams.js b/jstests/libs/override_methods/implicit_whole_db_changestreams.js
index 056a2659a0c..93f485e4564 100644
--- a/jstests/libs/override_methods/implicit_whole_db_changestreams.js
+++ b/jstests/libs/override_methods/implicit_whole_db_changestreams.js
@@ -99,6 +99,9 @@ const ChangeStreamPassthroughHelpers = {
     },
     upconvertGetMoreRequest: function(db, cmdObj) {
         return [this.execDBName(db), Object.assign({}, cmdObj, {collection: "$cmd.aggregate"})];
+    },
+    passthroughType: function() {
+        return ChangeStreamWatchMode.kDb;
     }
 };
 
diff --git a/jstests/noPassthrough/change_stream_failover.js b/jstests/noPassthrough/change_stream_failover.js
index 9929d3add13..46463ae99b2 100644
--- a/jstests/noPassthrough/change_stream_failover.js
+++ b/jstests/noPassthrough/change_stream_failover.js
@@ -17,8 +17,8 @@
 
     rst.initiate();
 
-    for (let key of Object.keys(ChangeStreamTest.WatchMode)) {
-        const watchMode = ChangeStreamTest.WatchMode[key];
+    for (let key of Object.keys(ChangeStreamWatchMode)) {
+        const watchMode = ChangeStreamWatchMode[key];
         jsTestLog("Running test for mode " + watchMode);
 
         const primary = rst.getPrimary();
diff --git a/jstests/sharding/change_stream_resume_from_different_mongos.js b/jstests/sharding/change_stream_resume_from_different_mongos.js
index 893bb47ab78..0ec7ab2d00c 100644
--- a/jstests/sharding/change_stream_resume_from_different_mongos.js
+++ b/jstests/sharding/change_stream_resume_from_different_mongos.js
@@ -17,8 +17,8 @@
         rs: {nodes: 3, setParameter: {periodicNoopIntervalSecs: 1, writePeriodicNoops: true}}
     });
 
-    for (let key of Object.keys(ChangeStreamTest.WatchMode)) {
-        const watchMode = ChangeStreamTest.WatchMode[key];
+    for (let key of Object.keys(ChangeStreamWatchMode)) {
+        const watchMode = ChangeStreamWatchMode[key];
         jsTestLog("Running test for mode " + watchMode);
 
         const s0DB = st.s0.getDB("test");
diff --git a/jstests/sharding/change_stream_shard_failover.js b/jstests/sharding/change_stream_shard_failover.js
index beb218f3584..ade85d27086 100644
--- a/jstests/sharding/change_stream_shard_failover.js
+++ b/jstests/sharding/change_stream_shard_failover.js
@@ -27,8 +27,8 @@ TestData.skipCheckingUUIDsConsistentAcrossCluster = true;
     const sDB = st.s.getDB("test");
     const kCollName = "change_stream_failover";
 
-    for (let key of Object.keys(ChangeStreamTest.WatchMode)) {
-        const watchMode = ChangeStreamTest.WatchMode[key];
+    for (let key of Object.keys(ChangeStreamWatchMode)) {
+        const watchMode = ChangeStreamWatchMode[key];
         jsTestLog("Running test for mode " + watchMode);
 
         const coll = assertDropAndRecreateCollection(sDB, kCollName);