SERVER-60926 Make enableSharding command optional

1 files changed, 22 insertions, 17 deletions

diff --git a/src/mongo/db/s/create_collection_coordinator.cpp b/src/mongo/db/s/create_collection_coordinator.cpp
index 4016982d6c1..72766e99945 100644
--- a/src/mongo/db/s/create_collection_coordinator.cpp
+++ b/src/mongo/db/s/create_collection_coordinator.cpp
@@ -57,6 +57,7 @@
 #include "mongo/s/cluster_commands_helpers.h"
 #include "mongo/s/cluster_write.h"
 #include "mongo/s/grid.h"
+#include "mongo/s/sharding_feature_flags_gen.h"
 
 namespace mongo {
 namespace {
@@ -630,25 +631,29 @@ ExecutorFuture<void> CreateCollectionCoordinator::_runImpl(
 void CreateCollectionCoordinator::_checkCommandArguments(OperationContext* opCtx) {
     LOGV2_DEBUG(5277902, 2, "Create collection _checkCommandArguments", "namespace"_attr = nss());
 
-    const auto dbEnabledForSharding = [&, this] {
-        // The modification of the 'sharded' flag for the db does not imply a database version
-        // change so we can't use the DatabaseShardingState to look it up. Instead we will do a
-        // first attempt through the catalog cache and if it is unset we will attempt another time
-        // after a forced catalog cache refresh.
-        auto catalogCache = Grid::get(opCtx)->catalogCache();
-
-        auto dbInfo = uassertStatusOK(catalogCache->getDatabase(opCtx, nss().db()));
-        if (!dbInfo.shardingEnabled()) {
-            sharding_ddl_util::linearizeCSRSReads(opCtx);
-            dbInfo = uassertStatusOK(catalogCache->getDatabaseWithRefresh(opCtx, nss().db()));
-        }
+    if (!feature_flags::gEnableShardingOptional.isEnabled(
+            serverGlobalParams.featureCompatibility)) {
 
-        return dbInfo.shardingEnabled();
-    }();
+        const auto dbEnabledForSharding = [&, this] {
+            // The modification of the 'sharded' flag for the db does not imply a database version
+            // change so we can't use the DatabaseShardingState to look it up. Instead we will do a
+            // first attempt through the catalog cache and if it is unset we will attempt another
+            // time after a forced catalog cache refresh.
+            auto catalogCache = Grid::get(opCtx)->catalogCache();
 
-    uassert(ErrorCodes::IllegalOperation,
-            str::stream() << "sharding not enabled for db " << nss().db(),
-            dbEnabledForSharding);
+            auto dbInfo = uassertStatusOK(catalogCache->getDatabase(opCtx, nss().db()));
+            if (!dbInfo.shardingEnabled()) {
+                sharding_ddl_util::linearizeCSRSReads(opCtx);
+                dbInfo = uassertStatusOK(catalogCache->getDatabaseWithRefresh(opCtx, nss().db()));
+            }
+
+            return dbInfo.shardingEnabled();
+        }();
+
+        uassert(ErrorCodes::IllegalOperation,
+                str::stream() << "sharding not enabled for db " << nss().db(),
+                dbEnabledForSharding);
+    }
 
     uassert(ErrorCodes::InvalidNamespace,
             str::stream() << "Namespace too long. Namespace: " << nss()