SERVER-30722 Make changestreams require FCV 3.6

This reverts commit 883c6c2.

Author: mtrussotto

jstests/noPassthrough/change_stream_feature_compatibility_version.js

@@ -0,0 +1,27 @@
+// Test that $changeStreams usage is disallowed when the featureCompatibilityVersion is 3.4.
+(function() {
+    "use strict";
+
+    const rst = new ReplSetTest({nodes: 1});
+    rst.startSet();
+    rst.initiate();
+    const conn = rst.getPrimary();
+
+    const testDB = conn.getDB("changeStreams_feature_compatibility_version");
+    const coll = testDB.coll;
+    assert.commandWorked(testDB.createCollection(coll.getName()));
+
+    // Make sure $changeStreams works with (default) featureCompatibilityVersion 3.6
+
+    assert.commandWorked(testDB.runCommand(
+        {aggregate: coll.getName(), pipeline: [{$changeStream: {}}], cursor: {}}));
+
+    // $changeStreams is not permitted when the featureCompatibilityVersion is 3.4.
+
+    assert.commandWorked(testDB.adminCommand({setFeatureCompatibilityVersion: "3.4"}));
+    assert.commandFailedWithCode(
+        testDB.runCommand({aggregate: coll.getName(), pipeline: [{$changeStream: {}}], cursor: {}}),
+        ErrorCodes.InvalidOptions);
+
+    rst.stopSet();
+})();

jstests/noPassthrough/unsupported_change_stream_deployments.js

@@ -21,8 +21,9 @@
     const masterSlaveFixture = new ReplTest("change_stream");
     const master = masterSlaveFixture.start(true);
     assertChangeStreamNotSupportedOnConnection(master);
-    const slave = masterSlaveFixture.start(false);
-    assertChangeStreamNotSupportedOnConnection(slave);
+    // Slaves start in the wrong FCV, (SERVER-31218) resulting in the wrong error code.
+    // const slave = masterSlaveFixture.start(false);
+    // assertChangeStreamNotSupportedOnConnection(slave);
 
     // Test a sharded cluster with standalone shards.
     const clusterWithStandalones = new ShardingTest({shards: 2});

src/mongo/db/pipeline/document_source_change_stream.cpp

@@ -33,6 +33,7 @@
 #include "mongo/bson/simple_bsonelement_comparator.h"
 #include "mongo/db/bson/bson_helper.h"
 #include "mongo/db/catalog/uuid_catalog.h"
+#include "mongo/db/commands/feature_compatibility_version_command_parser.h"
 #include "mongo/db/pipeline/close_change_stream_exception.h"
 #include "mongo/db/pipeline/document_source_check_resume_token.h"
 #include "mongo/db/pipeline/document_source_limit.h"
@@ -231,6 +232,14 @@ BSONObj DocumentSourceChangeStream::buildMatchFilter(const NamespaceString& nss,
 
 list<intrusive_ptr<DocumentSource>> DocumentSourceChangeStream::createFromBson(
     BSONElement elem, const intrusive_ptr<ExpressionContext>& expCtx) {
+    uassert(
+        ErrorCodes::InvalidOptions,
+        str::stream()
+            << "The featureCompatibilityVersion must be 3.6 to use the $changeStream stage. See "
+            << feature_compatibility_version::kDochubLink
+            << ".",
+        serverGlobalParams.featureCompatibility.version.load() !=
+            ServerGlobalParams::FeatureCompatibility::Version::k34);
     // TODO: Add sharding support here (SERVER-29141).
     uassert(
         40470, "The $changeStream stage is not supported on sharded systems.", !expCtx->inMongos);

src/mongo/db/pipeline/document_source_change_stream_test.cpp

@@ -68,11 +68,34 @@ static const Timestamp ts(100, 1);
 static const repl::OpTime optime(ts, 1);
 static const NamespaceString nss("unittests.change_stream");
 
-using ChangeStreamStageTestNoSetup = AggregationContextFixture;
+class EnsureFCV {
+public:
+    using Version = ServerGlobalParams::FeatureCompatibility::Version;
+    EnsureFCV(Version version)
+        : _origVersion(serverGlobalParams.featureCompatibility.version.load()) {
+        serverGlobalParams.featureCompatibility.version.store(version);
+    }
+    ~EnsureFCV() {
+        serverGlobalParams.featureCompatibility.version.store(_origVersion);
+    }
+
+private:
+    const Version _origVersion;
+};
+
+class ChangeStreamStageTestNoSetup : public AggregationContextFixture {
+public:
+    ChangeStreamStageTestNoSetup() : ChangeStreamStageTestNoSetup(nss) {}
+    ChangeStreamStageTestNoSetup(NamespaceString nsString)
+        : AggregationContextFixture(nsString), _ensureFCV(EnsureFCV::Version::k36) {}
+
+private:
+    EnsureFCV _ensureFCV;
+};
 
-class ChangeStreamStageTest : public AggregationContextFixture {
+class ChangeStreamStageTest : public ChangeStreamStageTestNoSetup {
 public:
-    ChangeStreamStageTest() : AggregationContextFixture(nss) {
+    ChangeStreamStageTest() : ChangeStreamStageTestNoSetup() {
         repl::ReplicationCoordinator::set(getExpCtx()->opCtx->getServiceContext(),
                                           stdx::make_unique<repl::ReplicationCoordinatorMock>(
                                               getExpCtx()->opCtx->getServiceContext()));

src/mongo/db/pipeline/pipeline_test.cpp

@@ -65,6 +65,21 @@ void setMockReplicationCoordinatorOnOpCtx(OperationContext* opCtx) {
         opCtx->getServiceContext(),
         stdx::make_unique<repl::ReplicationCoordinatorMock>(opCtx->getServiceContext()));
 }
+
+class EnsureFCV {
+public:
+    using Version = ServerGlobalParams::FeatureCompatibility::Version;
+    EnsureFCV(Version version)
+        : _origVersion(serverGlobalParams.featureCompatibility.version.load()) {
+        serverGlobalParams.featureCompatibility.version.store(version);
+    }
+    ~EnsureFCV() {
+        serverGlobalParams.featureCompatibility.version.store(_origVersion);
+    }
+
+private:
+    const Version _origVersion;
+};
 }  // namespace
 
 namespace Optimizations {
@@ -979,6 +994,7 @@ TEST(PipelineOptimizationTest, MatchOnMaxLengthShouldMoveAcrossRename) {
 }
 
 TEST(PipelineOptimizationTest, ChangeStreamLookupSwapsWithIndependentMatch) {
+    EnsureFCV ensureFCV(EnsureFCV::Version::k36);
     QueryTestServiceContext testServiceContext;
     auto opCtx = testServiceContext.makeOperationContext();
 
@@ -1004,6 +1020,7 @@ TEST(PipelineOptimizationTest, ChangeStreamLookupSwapsWithIndependentMatch) {
 }
 
 TEST(PipelineOptimizationTest, ChangeStreamLookupDoesNotSwapWithMatchOnPostImage) {
+    EnsureFCV ensureFCV(EnsureFCV::Version::k36);
     QueryTestServiceContext testServiceContext;
     auto opCtx = testServiceContext.makeOperationContext();
 
@@ -1482,6 +1499,7 @@ TEST_F(PipelineInitialSourceNSTest, AggregateOneNSValidForFacetPipelineRegardles
 }
 
 TEST_F(PipelineInitialSourceNSTest, ChangeStreamIsValidAsFirstStage) {
+    EnsureFCV ensureFCV(EnsureFCV::Version::k36);
     const std::vector<BSONObj> rawPipeline = {fromjson("{$changeStream: {}}")};
     auto ctx = getExpCtx();
     setMockReplicationCoordinatorOnOpCtx(ctx->opCtx);
@@ -1490,6 +1508,7 @@ TEST_F(PipelineInitialSourceNSTest, ChangeStreamIsValidAsFirstStage) {
 }
 
 TEST_F(PipelineInitialSourceNSTest, ChangeStreamIsNotValidIfNotFirstStage) {
+    EnsureFCV ensureFCV(EnsureFCV::Version::k36);
     const std::vector<BSONObj> rawPipeline = {fromjson("{$match: {custom: 'filter'}}"),
                                               fromjson("{$changeStream: {}}")};
     auto ctx = getExpCtx();
@@ -1500,6 +1519,7 @@ TEST_F(PipelineInitialSourceNSTest, ChangeStreamIsNotValidIfNotFirstStage) {
 }
 
 TEST_F(PipelineInitialSourceNSTest, ChangeStreamIsNotValidIfNotFirstStageInFacet) {
+    EnsureFCV ensureFCV(EnsureFCV::Version::k36);
     const std::vector<BSONObj> rawPipeline = {fromjson("{$match: {custom: 'filter'}}"),
                                               fromjson("{$changeStream: {}}")};
     auto ctx = getExpCtx();