SERVER-70704 Enable initial sync JS test for change collections.

2 files changed, 54 insertions, 37 deletions

diff --git a/jstests/serverless/initial_sync_change_collection.js b/jstests/serverless/initial_sync_change_collection.js
index 83aa6ac0820..ec5fdd006de 100644
--- a/jstests/serverless/initial_sync_change_collection.js
+++ b/jstests/serverless/initial_sync_change_collection.js
@@ -3,35 +3,34 @@
 // in the new secondary.
 // @tags: [
 //   requires_fcv_62,
-//   __TEMPORARILY_DISABLED__
 // ]
 //
 (function() {
 "use strict";
 
-load("jstests/libs/fail_point_util.js");                    // For waitForFailPoint.
-load("jstests/serverless/libs/change_collection_util.js");  // For verifyChangeCollectionEntries.
+// For waitForFailPoint.
+load("jstests/libs/fail_point_util.js");
+// For verifyChangeCollectionEntries and ChangeStreamMultitenantReplicaSetTest.
+load("jstests/serverless/libs/change_collection_util.js");
 
-const replSetTest = new ReplSetTest({nodes: 1});
-
-// TODO SERVER-70704 Remove '__TEMPORARILY_DISABLED__  tag and replace 'ReplSetTest' with
-// 'ChangeStreamMultitenantReplicaSetTest'.
-replSetTest.startSet({
-    setParameter: {
-        featureFlagServerlessChangeStreams: true,
-        multitenancySupport: true,
-        featureFlagMongoStore: true
-    }
-});
+const replSetTest = new ChangeStreamMultitenantReplicaSetTest({nodes: 1});
+const primary = replSetTest.getPrimary();
 
-replSetTest.initiate();
+// User id and the associated tenant id.
+const userInfo = {
+    tenantId: ObjectId(),
+    user: ObjectId().str
+};
 
-const primary = replSetTest.getPrimary();
+// Create a connection to the primary node for the tenant.
+const tenantPrimaryNode = ChangeStreamMultitenantReplicaSetTest.getTenantConnection(
+    primary.host, userInfo.tenantId, userInfo.user);
 
-// Enable the change stream to create the change collection.
-assert.commandWorked(primary.getDB("admin").runCommand({setChangeStreamState: 1, enabled: true}));
+// Enable the change stream for the tenant.
+replSetTest.setChangeStreamState(tenantPrimaryNode, true);
 
-const primaryChangeColl = primary.getDB("config").system.change_collection;
+// Get the change collection on the primary node for the tenant.
+const primaryChangeColl = tenantPrimaryNode.getDB("config").system.change_collection;
 
 const mdbStockPriceDoc = {
     _id: "mdb",
@@ -40,22 +39,17 @@ const mdbStockPriceDoc = {
 
 // The document 'mdbStockPriceDoc' is inserted before starting the initial sync. As such the
 // document 'mdbStockPriceDoc' should not be cloned in the secondary after initial sync is complete.
-assert.commandWorked(primary.getDB("test").stockPrice.insert(mdbStockPriceDoc));
+assert.commandWorked(tenantPrimaryNode.getDB("test").stockPrice.insert(mdbStockPriceDoc));
 assert.eq(primaryChangeColl.find({o: mdbStockPriceDoc}).toArray().length, 1);
 
 // Add a new secondary to the replica set and block the initial sync after the data cloning is done.
-const secondary = replSetTest.add({
+const secondary = replSetTest.addNode({
     setParameter: {
         // Hang after the data cloning phase is completed.
-        "failpoint.initialSyncHangAfterDataCloning": tojson({mode: "alwaysOn"}),
-        featureFlagServerlessChangeStreams: true,
-        multitenancySupport: true,
-        featureFlagMongoStore: true
+        "failpoint.initialSyncHangAfterDataCloning": tojson({mode: "alwaysOn"})
     }
 });
 
-replSetTest.reInitiate();
-
 // Wait for the cloning phase to complete. The cloning phase should not clone documents of the
 // change collection from the primary.
 assert.commandWorked(secondary.adminCommand({
@@ -71,7 +65,7 @@ const tslaStockPriceDoc = {
 
 // The document 'tslaStockPriceDoc' is inserted in the primary after the data cloning phase has
 // completed, as such this should be inserted in the secondary's change change collection.
-assert.commandWorked(primary.getDB("test").stockPrice.insert(tslaStockPriceDoc));
+assert.commandWorked(tenantPrimaryNode.getDB("test").stockPrice.insert(tslaStockPriceDoc));
 assert.eq(primaryChangeColl.find({o: tslaStockPriceDoc}).toArray().length, 1);
 
 // Unblock the initial sync process.
@@ -81,10 +75,14 @@ assert.commandWorked(secondary.getDB("test").adminCommand(
 // Wait for the initial sync to complete.
 replSetTest.waitForState(secondary, ReplSetTest.State.SECONDARY);
 
+// Create a connection to the secondary node for the tenant.
+const tenantSecondaryNode = ChangeStreamMultitenantReplicaSetTest.getTenantConnection(
+    secondary.host, userInfo.tenantId, userInfo.user);
+
 // Verify that the document 'mdbStockPriceDoc' does not exist and the document 'tslaStockPriceDoc'
 // exists in the secondary's change collection.
 const changeCollDocs =
-    secondary.getDB("config")
+    tenantSecondaryNode.getDB("config")
         .system.change_collection.find({$or: [{o: mdbStockPriceDoc}, {o: tslaStockPriceDoc}]})
         .toArray();
 assert.eq(changeCollDocs.length, 1);
@@ -98,9 +96,9 @@ const endOplogTimestamp = oplogDocs.at(-1).ts;
 
 // The change collection gets created at the data cloning phase and documents are written to the
 // oplog only after the data cloning is done. And so, the change collection already exists in place
-// to capture all oplog entries. As such, the change collection entries and the oplog entries from
-// the 'startOplogTimestamp' to the 'endOplogTimestamp' must be exactly the same.
-verifyChangeCollectionEntries(secondary, startOplogTimestamp, endOplogTimestamp);
+// to capture oplog entries. As such, the change collection entries and the oplog entries for
+// timestamp range ('startOplogTimestamp', 'endOplogTimestamp'] must be the same.
+verifyChangeCollectionEntries(secondary, startOplogTimestamp, endOplogTimestamp, userInfo.tenantId);
 
 // The state of the change collection after the initial sync is not consistent with the primary.
 // This is because the change collection's data is never cloned to the secondary, only it's creation
diff --git a/jstests/serverless/libs/change_collection_util.js b/jstests/serverless/libs/change_collection_util.js
index ad06fb41f4d..85d5ab20c4a 100644
--- a/jstests/serverless/libs/change_collection_util.js
+++ b/jstests/serverless/libs/change_collection_util.js
@@ -69,17 +69,21 @@ function verifyChangeCollectionEntries(
 class ChangeStreamMultitenantReplicaSetTest extends ReplSetTest {
     constructor(config) {
         // Instantiate the 'ReplSetTest' with 'serverless' as an option.
-        const newConfig = Object.assign(
+        const nodeConfig = Object.assign(
             {name: "ChangeStreamMultitenantReplicaSetTest", serverless: true}, config);
-        super(newConfig);
+        super(nodeConfig);
 
-        // Start and initialize the replica set.
-        const setParameter = Object.assign({}, config.setParameter || {}, {
+        // A dictionary of parameters required for multitenancy.
+        this._multitenancyParameters = {
             featureFlagServerlessChangeStreams: true,
             multitenancySupport: true,
             featureFlagMongoStore: true,
             featureFlagRequireTenantID: true
-        });
+        };
+
+        // Start and initialize the replica set.
+        const setParameter =
+            Object.assign({}, config.setParameter || {}, this._multitenancyParameters);
         this.startSet({setParameter: setParameter});
         this.initiate();
 
@@ -89,6 +93,21 @@ class ChangeStreamMultitenantReplicaSetTest extends ReplSetTest {
             {createUser: "root", pwd: "pwd", roles: ["root"]}));
     }
 
+    // Adds a node to the replica set with the provided configuration 'config'.
+    addNode(config) {
+        // Get the a copy of the 'config' dictionary and add required multitenancy flags to it.
+        const nodeConfig = Object.assign({serverless: true}, config);
+        nodeConfig.setParameter =
+            Object.assign({}, nodeConfig.setParameter || {}, this._multitenancyParameters);
+
+        // Initiate the replica set with the newly added node.
+        const node = this.add(nodeConfig);
+        this.reInitiate();
+
+        // Return the newly added node.
+        return node;
+    }
+
     // Returns a connection to the 'hostAddr' with 'tenantId' stamped to it for the created user.
     static getTenantConnection(hostAddr,
                                tenantId,