SERVER-66717 Guard change collection code with the serverless flag.

13 files changed, 71 insertions, 45 deletions

diff --git a/buildscripts/resmokeconfig/fully_disabled_feature_flags.yml b/buildscripts/resmokeconfig/fully_disabled_feature_flags.yml
index 3c43a042097..678b3c522c3 100644
--- a/buildscripts/resmokeconfig/fully_disabled_feature_flags.yml
+++ b/buildscripts/resmokeconfig/fully_disabled_feature_flags.yml
@@ -12,9 +12,5 @@
 # released create the transactions collection index and is only meant to be enabled adhoc, so only
 # its targeted tests should enable it.
 - featureFlagAlwaysCreateConfigTransactionsPartialIndexOnStepUp
-# Disable 'featureFlagServerlessChangeStreams' until the change collection becomes stable.
-# TODO SERVER-67267 remove the flag.
-- featureFlagServerlessChangeStreams
-
 # Disable 'featureFlagRangeDeleterService' as long as the service is not performing actual range deletions
 - featureFlagRangeDeleterService
diff --git a/buildscripts/resmokeconfig/suites/change_streams_multitenant_passthrough.yml b/buildscripts/resmokeconfig/suites/change_streams_multitenant_passthrough.yml
index 16a0397b526..47e9179a50b 100644
--- a/buildscripts/resmokeconfig/suites/change_streams_multitenant_passthrough.yml
+++ b/buildscripts/resmokeconfig/suites/change_streams_multitenant_passthrough.yml
@@ -50,8 +50,10 @@ executor:
   - class: EnableChangeStream
   fixture:
     class: ReplicaSetFixture
+    replset_name: "ChangeStreamMultitenantReplSet"
     mongod_options:
       bind_ip_all: ''
+      serverless: true
       set_parameters:
         enableTestCommands: 1
         featureFlagServerlessChangeStreams: true
diff --git a/jstests/auth/change_stream_change_collection_role_auth.js b/jstests/auth/change_stream_change_collection_role_auth.js
index 60b86a8d823..384bdbb5b66 100644
--- a/jstests/auth/change_stream_change_collection_role_auth.js
+++ b/jstests/auth/change_stream_change_collection_role_auth.js
@@ -157,10 +157,14 @@ function removeChangeCollectionDoc(authDb) {
 
 // Start a replica-set test with one-node and authentication enabled. Connect to the primary node
 // and create users.
-const replSetTest =
-    new ReplSetTest({name: "shard", nodes: 1, useHostName: true, waitForKeys: false});
+const replSetTest = new ReplSetTest({
+    name: "shard",
+    nodes: 1,
+    useHostName: true,
+    waitForKeys: false,
+    serverless: true,
+});
 
-// TODO SERVER-67267: Add 'serverless' flags.
 replSetTest.startSet({
     keyFile: keyFile,
     setParameter: {
diff --git a/jstests/libs/override_methods/override_fixtures_changestream_multitenancy.js b/jstests/libs/override_methods/override_fixtures_changestream_multitenancy.js
index c9b713bda0c..0ad1b36fc67 100644
--- a/jstests/libs/override_methods/override_fixtures_changestream_multitenancy.js
+++ b/jstests/libs/override_methods/override_fixtures_changestream_multitenancy.js
@@ -7,9 +7,15 @@
 const originalReplSet = ReplSetTest;
 
 ReplSetTest = function(opts) {
+    // Setup the 'serverless' environment if the 'opts' is not a connection string, ie. the
+    // replica-set does not already exist and the replica-set is not part of the sharded cluster,
+    // ie. 'setParametersMongos' property does not exist.
+    const newOpts = typeof opts !== "string" && !TestData.hasOwnProperty("setParametersMongos")
+        ? Object.assign({name: "OverridenServerlessChangeStreamReplSet", serverless: true}, opts)
+        : opts;
+
     // Call the constructor with the original 'ReplSetTest' to populate 'this' with required fields.
-    // TODO SERVER-67267 add {serverless:true} to the 'opts'.
-    originalReplSet.apply(this, [opts]);
+    originalReplSet.apply(this, [newOpts]);
 
     // Make a copy of the original 'startSetAsync' function and then override it to include the
     // required parameters.
diff --git a/jstests/serverless/change_stream_state_commands.js b/jstests/serverless/change_stream_state_commands.js
index 373bad342b5..f596953a39c 100644
--- a/jstests/serverless/change_stream_state_commands.js
+++ b/jstests/serverless/change_stream_state_commands.js
@@ -12,10 +12,10 @@ load('jstests/libs/parallel_shell_helpers.js');  // For funWithArgs.
 
 const replSetTest = new ReplSetTest({nodes: 2});
 
-// TODO SERVER-67267 Add 'serverless' flag.
-// TODO SERVER-68947 Add 'featureFlagRequireTenantID' flag.
-// TODO SERVER-69115 Remove '__TEMPORARILY_DISABLED__'
+// TODO SERVER-69115 Remove '__TEMPORARILY_DISABLED__ and use
+// 'ChangeStreamMultitenantReplicaSetTest'.
 replSetTest.startSet({
+    serverless: true,
     setParameter: {
         featureFlagServerlessChangeStreams: true,
         multitenancySupport: true,
diff --git a/jstests/serverless/initial_sync_change_collection.js b/jstests/serverless/initial_sync_change_collection.js
index 3d111307dd9..0fc4cba31d8 100644
--- a/jstests/serverless/initial_sync_change_collection.js
+++ b/jstests/serverless/initial_sync_change_collection.js
@@ -14,9 +14,8 @@ load("jstests/serverless/libs/change_collection_util.js");  // For verifyChangeC
 
 const replSetTest = new ReplSetTest({nodes: 1});
 
-// TODO SERVER-67267 Add 'serverless' flag.
-// TODO SERVER-69115 Add 'featureFlagRequireTenantID' flag and remove '__TEMPORARILY_DISABLED__'
-// tag and replace 'ReplSetTest' with 'ChangeStreamMultitenantReplicaSetTest'.
+// TODO SERVER-69115 Remove '__TEMPORARILY_DISABLED__  tag and replace 'ReplSetTest' with
+// 'ChangeStreamMultitenantReplicaSetTest'.
 replSetTest.startSet({
     setParameter: {
         featureFlagServerlessChangeStreams: true,
diff --git a/jstests/serverless/libs/change_collection_util.js b/jstests/serverless/libs/change_collection_util.js
index f9b8a9c6846..ae3370efb2a 100644
--- a/jstests/serverless/libs/change_collection_util.js
+++ b/jstests/serverless/libs/change_collection_util.js
@@ -66,11 +66,12 @@ function verifyChangeCollectionEntries(
 // This class also provides helpers that are commonly used when working with change collections.
 class ChangeStreamMultitenantReplicaSetTest extends ReplSetTest {
     constructor(config) {
-        // Instantiate the 'ReplSetTest'.
-        super(config);
+        // Instantiate the 'ReplSetTest' with 'serverless' as an option.
+        const newConfig = Object.assign(
+            {name: "ChangeStreamMultitenantReplicaSetTest", serverless: true}, config);
+        super(newConfig);
 
         // Start and initialize the replica set.
-        // TODO SERVER-67267 Add 'serverless' flag.
         const setParameter = Object.assign({}, config.setParameter || {}, {
             featureFlagServerlessChangeStreams: true,
             multitenancySupport: true,
diff --git a/jstests/serverless/write_to_change_collection_in_startup_recovery.js b/jstests/serverless/write_to_change_collection_in_startup_recovery.js
index bbf7cd0a259..147de958225 100644
--- a/jstests/serverless/write_to_change_collection_in_startup_recovery.js
+++ b/jstests/serverless/write_to_change_collection_in_startup_recovery.js
@@ -13,9 +13,8 @@ load("jstests/serverless/libs/change_collection_util.js");  // For verifyChangeC
 
 const replSetTest = new ReplSetTest({nodes: 1});
 
-// TODO SERVER-67267 Add 'serverless' flag.
-// TODO SERVER-69115 Add 'featureFlagRequireTenantID' flag and remove '__TEMPORARILY_DISABLED__'
-// tag and replace 'ReplSetTest' with 'ChangeStreamMultitenantReplicaSetTest'.
+// TODO SERVER-69115 Remove '__TEMPORARILY_DISABLED__ tag and replace 'ReplSetTest' with
+// 'ChangeStreamMultitenantReplicaSetTest'.
 replSetTest.startSet({
     setParameter: {
         featureFlagServerlessChangeStreams: true,
diff --git a/src/mongo/db/SConscript b/src/mongo/db/SConscript
index be4f755fac0..f5ee9f3988f 100644
--- a/src/mongo/db/SConscript
+++ b/src/mongo/db/SConscript
@@ -555,6 +555,7 @@ env.Library(
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/catalog/collection',
         '$BUILD_DIR/mongo/db/catalog/collection_catalog',
+        '$BUILD_DIR/mongo/db/global_settings',
         '$BUILD_DIR/mongo/db/query/query_knobs',
         '$BUILD_DIR/mongo/db/server_base',
         '$BUILD_DIR/mongo/db/server_options',
diff --git a/src/mongo/db/change_collection_expired_documents_remover.cpp b/src/mongo/db/change_collection_expired_documents_remover.cpp
index d84fb21b672..ce77d924b99 100644
--- a/src/mongo/db/change_collection_expired_documents_remover.cpp
+++ b/src/mongo/db/change_collection_expired_documents_remover.cpp
@@ -78,13 +78,6 @@ boost::optional<int64_t> getExpireAfterSeconds(const TenantId& tenantId) {
 }
 
 void removeExpiredDocuments(Client* client) {
-    // TODO SERVER-66717 Remove this logic from this method. Due to the delay in the feature flag
-    // activation it was placed here. The remover job should ultimately be initialized at the mongod
-    // startup when launched in serverless mode.
-    if (!change_stream_serverless_helpers::isChangeCollectionsModeActive()) {
-        return;
-    }
-
     hangBeforeRemovingExpiredChanges.pauseWhileSet();
 
     try {
@@ -211,12 +204,16 @@ private:
 }  // namespace
 
 void startChangeCollectionExpiredDocumentsRemover(ServiceContext* serviceContext) {
-    LOGV2(6663507, "Starting the ChangeCollectionExpiredChangeRemover");
-    ChangeCollectionExpiredDocumentsRemover::start(serviceContext);
+    if (change_stream_serverless_helpers::canInitializeServices()) {
+        LOGV2(6663507, "Starting the ChangeCollectionExpiredChangeRemover");
+        ChangeCollectionExpiredDocumentsRemover::start(serviceContext);
+    }
 }
 
 void shutdownChangeCollectionExpiredDocumentsRemover(ServiceContext* serviceContext) {
-    LOGV2(6663508, "Shutting down the ChangeCollectionExpiredChangeRemover");
-    ChangeCollectionExpiredDocumentsRemover::shutdown(serviceContext);
+    if (change_stream_serverless_helpers::canInitializeServices()) {
+        LOGV2(6663508, "Shutting down the ChangeCollectionExpiredChangeRemover");
+        ChangeCollectionExpiredDocumentsRemover::shutdown(serviceContext);
+    }
 }
 }  // namespace mongo
diff --git a/src/mongo/db/change_stream_serverless_helpers.cpp b/src/mongo/db/change_stream_serverless_helpers.cpp
index 0577894e397..b5578f35246 100644
--- a/src/mongo/db/change_stream_serverless_helpers.cpp
+++ b/src/mongo/db/change_stream_serverless_helpers.cpp
@@ -33,6 +33,8 @@
 
 #include "mongo/db/catalog_raii.h"
 #include "mongo/db/dbdirectclient.h"
+#include "mongo/db/global_settings.h"
+#include "mongo/db/multitenancy_gen.h"
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/server_feature_flags_gen.h"
 #include "mongo/db/server_options.h"
@@ -41,13 +43,10 @@ namespace mongo {
 namespace change_stream_serverless_helpers {
 
 bool isChangeCollectionsModeActive() {
-    // A change collection must not be enabled on the config server.
-    if (serverGlobalParams.clusterRole == ClusterRole::ConfigServer) {
-        return false;
-    }
-
-    // TODO SERVER-67267 guard with 'multitenancySupport' and 'isServerless' flag.
-    return serverGlobalParams.featureCompatibility.isVersionInitialized() &&
+    // A change collection mode is declared as active if the required services can be initialized,
+    // the feature flag is enabled and the FCV version is already initialized.
+    return canInitializeServices() &&
+        serverGlobalParams.featureCompatibility.isVersionInitialized() &&
         feature_flags::gFeatureFlagServerlessChangeStreams.isEnabled(
             serverGlobalParams.featureCompatibility);
 }
@@ -65,6 +64,20 @@ bool isChangeStreamEnabled(OperationContext* opCtx, const TenantId& tenantId) {
             opCtx, NamespaceString::makePreImageCollectionNSS(boost::none)));
 }
 
+bool canInitializeServices() {
+    // A change collection must not be enabled on the config server.
+    if (serverGlobalParams.clusterRole == ClusterRole::ConfigServer) {
+        return false;
+    }
+
+    // A change stream services are enabled only in the multitenant serverless settings. For the
+    // sharded cluster, 'internalChangeStreamUseTenantIdForTesting' maybe provided for the testing
+    // purposes until the support is available.
+    const auto isMultiTenantServerless =
+        getGlobalReplSettings().isServerless() && gMultitenancySupport;
+    return isMultiTenantServerless || internalChangeStreamUseTenantIdForTesting.load();
+}
+
 const TenantId& getTenantIdForTesting() {
     static const TenantId kTestTenantId(
         OID("00000000"   /* timestamp */
diff --git a/src/mongo/db/change_stream_serverless_helpers.h b/src/mongo/db/change_stream_serverless_helpers.h
index bdeb04f3ff7..f3f63cc97f8 100644
--- a/src/mongo/db/change_stream_serverless_helpers.h
+++ b/src/mongo/db/change_stream_serverless_helpers.h
@@ -51,6 +51,12 @@ bool isChangeCollectionsModeActive();
 bool isChangeStreamEnabled(OperationContext* opCtx, const TenantId& tenantId);
 
 /**
+ * Returns true if services related to the serverless change stream can be initialized.
+ * TODO SERVER-69960 Remove this function and use 'isChangeCollectionsModeActive' instead.
+ */
+bool canInitializeServices();
+
+/**
  * Returns an internal tenant id that will be used for testing purposes. This tenant id will not
  * conflict with any other tenant id.
  */
diff --git a/src/mongo/db/mongod_main.cpp b/src/mongo/db/mongod_main.cpp
index b344f20daf4..c8048e567f2 100644
--- a/src/mongo/db/mongod_main.cpp
+++ b/src/mongo/db/mongod_main.cpp
@@ -59,6 +59,7 @@
 #include "mongo/db/change_collection_expired_documents_remover.h"
 #include "mongo/db/change_stream_change_collection_manager.h"
 #include "mongo/db/change_stream_options_manager.h"
+#include "mongo/db/change_stream_serverless_helpers.h"
 #include "mongo/db/client.h"
 #include "mongo/db/client_metadata_propagation_egress_hook.h"
 #include "mongo/db/clientcursor.h"
@@ -349,8 +350,10 @@ void registerPrimaryOnlyServices(ServiceContext* serviceContext) {
         }
     }
 
-    // TODO SERVER-66717 create 'SetChangeStreamStateCoordinatorService' only in the serverless.
-    services.push_back(std::make_unique<SetChangeStreamStateCoordinatorService>(serviceContext));
+    if (change_stream_serverless_helpers::canInitializeServices()) {
+        services.push_back(
+            std::make_unique<SetChangeStreamStateCoordinatorService>(serviceContext));
+    }
 
     for (auto& service : services) {
         registry->registerService(std::move(service));
@@ -790,8 +793,6 @@ ExitCode _initAndListen(ServiceContext* serviceContext, int listenPort) {
         repl::ReplicationCoordinator::modeNone;
     if (!isStandalone) {
         startChangeStreamExpiredPreImagesRemover(serviceContext);
-        // TODO SERVER-66717 Start 'startChangeCollectionExpiredDocumentsRemover' only in the
-        // serverless.
         startChangeCollectionExpiredDocumentsRemover(serviceContext);
     }
 
@@ -1576,8 +1577,9 @@ int mongod_main(int argc, char* argv[]) {
     ReadWriteConcernDefaults::create(service, readWriteConcernDefaultsCacheLookupMongoD);
     ChangeStreamOptionsManager::create(service);
 
-    // TODO SERVER-66717 Create 'ChangeStreamChangeCollectionManager' only in the serverless.
-    ChangeStreamChangeCollectionManager::create(service);
+    if (change_stream_serverless_helpers::canInitializeServices()) {
+        ChangeStreamChangeCollectionManager::create(service);
+    }
 
 #if defined(_WIN32)
     if (ntservice::shouldStartService()) {