SERVER-69072 Add version restrictions to change stream expireAfterSeconds parameters

6 files changed, 96 insertions, 4 deletions

diff --git a/jstests/libs/cluster_server_parameter_utils.js b/jstests/libs/cluster_server_parameter_utils.js
index 62f041c1348..9ea692e9649 100644
--- a/jstests/libs/cluster_server_parameter_utils.js
+++ b/jstests/libs/cluster_server_parameter_utils.js
@@ -16,6 +16,8 @@
  *
  */
 
+load("jstests/libs/feature_flag_util.js");
+
 const testOnlyClusterParameterNames = [
     "testStrClusterParameter",
     "testIntClusterParameter",
@@ -24,6 +26,25 @@ const testOnlyClusterParameterNames = [
 const nonTestClusterParameterNames = ["changeStreamOptions", "changeStreams"];
 const clusterParameterNames = testOnlyClusterParameterNames.concat(nonTestClusterParameterNames);
 
+// A dictionary to ignore the cluster parameters based on specific criteria. The key is the name of
+// a cluster parameter and the value is a boolean function returning 'true' in cases when the
+// parameter should be ignored.
+const ignoreParametersDict = {
+    changeStreamOptions: function(conn) {
+        return FeatureFlagUtil.isEnabled(conn, "ServerlessChangeStreams");
+    },
+    changeStreams: function(conn) {
+        return !FeatureFlagUtil.isEnabled(conn, "ServerlessChangeStreams");
+    }
+};
+
+function ignoreParameter(paramName, conn) {
+    if (ignoreParametersDict[paramName]) {
+        return ignoreParametersDict[paramName](conn);
+    }
+    return false;
+}
+
 const testOnlyClusterParametersDefault = [
     {
         _id: "testStrClusterParameter",
@@ -131,6 +152,9 @@ function setupSharded(st) {
 // Upserts config.clusterParameters document with w:majority via setClusterParameter.
 function runSetClusterParameter(conn, update) {
     const paramName = update._id;
+    if (ignoreParameter(paramName, conn)) {
+        return;
+    }
     let updateCopy = Object.assign({}, update);
     delete updateCopy._id;
     delete updateCopy.clusterParameterTime;
@@ -171,6 +195,9 @@ function runGetClusterParameterNode(conn, getClusterParameterArgs, expectedClust
                 }
                 return sorted;
             }, {});
+        if (ignoreParameter(expectedClusterParameter._id, conn)) {
+            continue;
+        }
         if (bsonWoCompare(sortedExpectedClusterParameter, sortedActualClusterParameter) !== 0) {
             print('expected: ' + tojson(sortedExpectedClusterParameter) +
                   '\nactual: ' + tojson(sortedActualClusterParameter));
diff --git a/jstests/noPassthrough/change_stream_options.js b/jstests/noPassthrough/change_stream_options.js
index 2a36bb840dd..1e8a9828ba8 100644
--- a/jstests/noPassthrough/change_stream_options.js
+++ b/jstests/noPassthrough/change_stream_options.js
@@ -7,6 +7,9 @@
 (function() {
 "use strict";
 
+// For ChangeStreamMultitenantReplicaSetTest.
+load("jstests/serverless/libs/change_collection_util.js");
+
 const testDBName = jsTestName();
 
 // Tests set and get change stream options command with 'admin' database.
@@ -180,4 +183,20 @@ function testChangeStreamOptionsWithAdminDB(conn) {
 
     replSetTest.stopSet();
 })();
+
+// Tests that 'changeStreamOptions.preAndPostImages.expireAfterSeconds' is not available in
+// serverless.
+(function testChangeStreamOptionsInServerless() {
+    const replSetTest = new ChangeStreamMultitenantReplicaSetTest({nodes: 1});
+
+    const primary = replSetTest.getPrimary();
+    const adminDB = primary.getDB("admin");
+    assert.commandFailedWithCode(adminDB.runCommand({
+        setClusterParameter:
+            {changeStreamOptions: {preAndPostImages: {expireAfterSeconds: NumberLong(40)}}}
+    }),
+                                 ErrorCodes.CommandNotSupported);
+
+    replSetTest.stopSet();
+})();
 }());
diff --git a/jstests/noPassthrough/change_streams_cluster_parameter.js b/jstests/serverless/change_streams_cluster_parameter.js
index 6297c785cd0..c6e23058307 100644
--- a/jstests/noPassthrough/change_streams_cluster_parameter.js
+++ b/jstests/serverless/change_streams_cluster_parameter.js
@@ -10,6 +10,9 @@
 (function() {
 "use strict";
 
+// For ChangeStreamMultitenantReplicaSetTest.
+load("jstests/serverless/libs/change_collection_util.js");
+
 // Verifies that the 'getClusterParameter' on the 'changeStreams' cluster-wide parameter returns the
 // expected response.
 function assertGetResponse(db, expectedChangeStreamParam) {
@@ -69,9 +72,7 @@ function testWithoutAdminDB(conn) {
 
 // Tests the set and get change streams parameter on the replica-set.
 {
-    const rst = new ReplSetTest({name: "replSet", nodes: 2});
-    rst.startSet();
-    rst.initiate();
+    const rst = new ChangeStreamMultitenantReplicaSetTest({name: "replSet", nodes: 2});
 
     const primary = rst.getPrimary();
     const secondary = rst.getSecondaries()[0];
@@ -89,7 +90,14 @@ function testWithoutAdminDB(conn) {
 
 // Tests the set and get change streams parameter on the sharded cluster.
 {
-    const st = new ShardingTest({shards: 1, mongos: 1});
+    const st = new ShardingTest({
+        shards: 1,
+        mongos: 1,
+        other: {
+            mongosOptions: {setParameter: {featureFlagServerlessChangeStreams: true}},
+            shardOptions: {setParameter: {featureFlagServerlessChangeStreams: true}}
+        }
+    });
     const adminDB = st.rs0.getPrimary().getDB("admin");
 
     // Test that setClusterParameter cannot be issued directly on shards in the sharded cluster,
@@ -105,4 +113,21 @@ function testWithoutAdminDB(conn) {
 
     st.stop();
 }
+
+// Tests that 'changeStreams.expireAfterSeconds' is only available in serverless.
+{
+    const rst = new ReplSetTest({nodes: 1});
+    rst.startSet({setParameter: {featureFlagServerlessChangeStreams: false}});
+    rst.initiate();
+
+    const primary = rst.getPrimary();
+    const adminDB = primary.getDB("admin");
+
+    assert.commandFailedWithCode(
+        adminDB.runCommand(
+            {setClusterParameter: {changeStreams: {expireAfterSeconds: NumberLong(10)}}}),
+        ErrorCodes.CommandNotSupported);
+
+    rst.stopSet();
+}
 }());
diff --git a/src/mongo/db/SConscript b/src/mongo/db/SConscript
index fe62eceee81..39f8650777e 100644
--- a/src/mongo/db/SConscript
+++ b/src/mongo/db/SConscript
@@ -476,6 +476,7 @@ env.Library(
     ],
     LIBDEPS_PRIVATE=[
         'change_stream_options',
+        'change_stream_serverless_helpers',
         'service_context',
     ],
 )
@@ -488,7 +489,9 @@ env.Library(
     ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/idl/cluster_server_parameter',
+        'change_stream_serverless_helpers',
         'server_base',
+        'server_options_core',
     ],
 )
 
diff --git a/src/mongo/db/change_stream_options_manager.cpp b/src/mongo/db/change_stream_options_manager.cpp
index 541b8486621..a9a417aede6 100644
--- a/src/mongo/db/change_stream_options_manager.cpp
+++ b/src/mongo/db/change_stream_options_manager.cpp
@@ -32,6 +32,7 @@
 
 #include "mongo/db/change_stream_options_manager.h"
 #include "mongo/db/change_stream_options_parameter_gen.h"
+#include "mongo/db/change_stream_serverless_helpers.h"
 #include "mongo/logv2/log.h"
 
 #define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kCommand
@@ -131,6 +132,14 @@ Status ChangeStreamOptionsParameter::validate(const BSONElement& newValueElement
                     }
                 },
                 [&](const std::int64_t& expireAfterSeconds) {
+                    if (change_stream_serverless_helpers::isChangeCollectionsModeActive()) {
+                        validateStatus = {
+                            ErrorCodes::CommandNotSupported,
+                            "The 'changeStreamOptions.preAndPostImages.expireAfterSeconds' is "
+                            "unsupported in serverless, consider using "
+                            "'changeStreams.expireAfterSeconds' instead."};
+                        return;
+                    }
                     if (expireAfterSeconds <= 0) {
                         validateStatus = {
                             ErrorCodes::BadValue,
diff --git a/src/mongo/db/change_streams_cluster_parameter.cpp b/src/mongo/db/change_streams_cluster_parameter.cpp
index b4c798ccc1c..96e9df4c0a0 100644
--- a/src/mongo/db/change_streams_cluster_parameter.cpp
+++ b/src/mongo/db/change_streams_cluster_parameter.cpp
@@ -32,6 +32,7 @@
 #include "mongo/db/change_streams_cluster_parameter.h"
 
 #include "mongo/base/status.h"
+#include "mongo/db/change_stream_serverless_helpers.h"
 #include "mongo/db/change_streams_cluster_parameter_gen.h"
 #include "mongo/logv2/log.h"
 namespace mongo {
@@ -39,6 +40,14 @@ namespace mongo {
 Status validateChangeStreamsClusterParameter(
     const ChangeStreamsClusterParameterStorage& clusterParameter,
     const boost::optional<TenantId>& tenantId) {
+    // 'isChangeCollectionsModeActive' always returns false on a config server, however, setting
+    // 'changeStreams.expireAfterSeconds' parameter on mongos will also change it on config servers.
+    if (serverGlobalParams.clusterRole != ClusterRole::ConfigServer &&
+        !change_stream_serverless_helpers::isChangeCollectionsModeActive()) {
+        return Status(
+            ErrorCodes::CommandNotSupported,
+            "The 'changeStreams' cluster-wide parameter is only available in serverless.");
+    }
     if (clusterParameter.getExpireAfterSeconds() <= 0) {
         return Status(ErrorCodes::BadValue,
                       "Expected a positive integer for 'expireAfterSeconds' field");