Update upstream source from tag 'upstream/6.0.10'debian/6.0.10-1

Update to upstream version '6.0.10'
with Debian dir 2d176fa254eee97b139f712fec5709641335a8c3

18 files changed, 1126 insertions, 477 deletions

diff --git a/src/mongo/db/s/config/configsvr_commit_move_primary_command.cpp b/src/mongo/db/s/config/configsvr_commit_move_primary_command.cpp
new file mode 100644
index 00000000000..76fc3f7b5c9
--- /dev/null
+++ b/src/mongo/db/s/config/configsvr_commit_move_primary_command.cpp
@@ -0,0 +1,102 @@
+/**
+ *    Copyright (C) 2022-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kSharding
+
+#include "mongo/db/auth/authorization_session.h"
+#include "mongo/db/commands.h"
+#include "mongo/db/repl/read_concern_args.h"
+#include "mongo/db/s/config/sharding_catalog_manager.h"
+#include "mongo/s/request_types/move_primary_gen.h"
+
+namespace mongo {
+namespace {
+
+class ConfigsvrCommitMovePrimaryCommand final
+    : public TypedCommand<ConfigsvrCommitMovePrimaryCommand> {
+public:
+    using Request = ConfigsvrCommitMovePrimary;
+
+    class Invocation final : public InvocationBase {
+    public:
+        using InvocationBase::InvocationBase;
+
+        void typedRun(OperationContext* opCtx) {
+            uassert(ErrorCodes::IllegalOperation,
+                    str::stream() << Request::kCommandName << " can only be run on config servers",
+                    serverGlobalParams.clusterRole == ClusterRole::ConfigServer);
+
+            // Set the operation context read concern level to local for reads into the config
+            // database.
+            repl::ReadConcernArgs::get(opCtx) =
+                repl::ReadConcernArgs(repl::ReadConcernLevel::kLocalReadConcern);
+
+            ShardingCatalogManager::get(opCtx)->commitMovePrimary(
+                opCtx,
+                request().getCommandParameter(),
+                request().getExpectedDatabaseVersion(),
+                request().getTo());
+        }
+
+    private:
+        NamespaceString ns() const override {
+            return NamespaceString(request().getDbName());
+        }
+
+        bool supportsWriteConcern() const override {
+            return true;
+        }
+
+        void doCheckAuthorization(OperationContext* opCtx) const override {
+            uassert(ErrorCodes::Unauthorized,
+                    "Unauthorized",
+                    AuthorizationSession::get(opCtx->getClient())
+                        ->isAuthorizedForActionsOnResource(ResourcePattern::forClusterResource(),
+                                                           ActionType::internal));
+        }
+    };
+
+private:
+    std::string help() const override {
+        return "Reassign a new primary shard for the given database on the config server. This is "
+               "an internal command only invokable on the config server, therefore do not call "
+               "directly.";
+    }
+
+    bool adminOnly() const override {
+        return true;
+    }
+
+    AllowedOnSecondary secondaryAllowed(ServiceContext* context) const override {
+        return AllowedOnSecondary::kNever;
+    }
+} configsvrCommitMovePrimaryCommand;
+
+}  // namespace
+}  // namespace mongo
diff --git a/src/mongo/db/s/config/configsvr_configure_collection_balancing.cpp b/src/mongo/db/s/config/configsvr_configure_collection_balancing.cpp
index a9db28064b5..c471c70437f 100644
--- a/src/mongo/db/s/config/configsvr_configure_collection_balancing.cpp
+++ b/src/mongo/db/s/config/configsvr_configure_collection_balancing.cpp
@@ -36,6 +36,7 @@
 #include "mongo/db/auth/authorization_session.h"
 #include "mongo/db/auth/privilege.h"
 #include "mongo/db/commands.h"
+#include "mongo/db/commands/feature_compatibility_version.h"
 #include "mongo/db/operation_context.h"
 #include "mongo/db/repl/repl_client_info.h"
 #include "mongo/db/s/balancer/balancer.h"
@@ -60,6 +61,16 @@ public:
 
         void typedRun(OperationContext* opCtx) {
             opCtx->setAlwaysInterruptAtStepDownOrUp();
+
+            // Hold the FCV region to serialize with the setFeatureCompatibilityVersion command
+            FixedFCVRegion fcvRegion(opCtx);
+            uassert(ErrorCodes::IllegalOperation,
+                    "_configsvrConfigureCollectionBalancing can only be run when the cluster is in "
+                    "feature "
+                    "compatibility versions greater or equal than 5.3.",
+                    serverGlobalParams.featureCompatibility.isGreaterThanOrEqualTo(
+                        multiversion::FeatureCompatibilityVersion::kVersion_5_3));
+
             uassert(ErrorCodes::IllegalOperation,
                     str::stream() << Request::kCommandName << " can only be run on config servers",
                     serverGlobalParams.clusterRole == ClusterRole::ConfigServer);
diff --git a/src/mongo/db/s/config/configsvr_merge_chunks_command.cpp b/src/mongo/db/s/config/configsvr_merge_chunks_command.cpp
index ea2823dcdf0..5df553b5bd3 100644
--- a/src/mongo/db/s/config/configsvr_merge_chunks_command.cpp
+++ b/src/mongo/db/s/config/configsvr_merge_chunks_command.cpp
@@ -94,8 +94,7 @@ public:
                                                                       request().getTimestamp(),
                                                                       request().getCollectionUUID(),
                                                                       request().getChunkRange(),
-                                                                      request().getShard(),
-                                                                      request().getValidAfter()));
+                                                                      request().getShard()));
             return ConfigSvrMergeResponse{ChunkVersion::fromBSONPositionalOrNewerFormat(
                 shardAndCollVers[ChunkVersion::kShardVersionField])};
         }
diff --git a/src/mongo/db/s/config/configsvr_run_restore_command.cpp b/src/mongo/db/s/config/configsvr_run_restore_command.cpp
index 166c10a4e70..22e3f7b2985 100644
--- a/src/mongo/db/s/config/configsvr_run_restore_command.cpp
+++ b/src/mongo/db/s/config/configsvr_run_restore_command.cpp
@@ -73,6 +73,25 @@ ShouldRestoreDocument shouldRestoreDocument(OperationContext* opCtx,
                                                    : ShouldRestoreDocument::kNo;
 }
 
+std::set<std::string> getDatabasesToRestore(OperationContext* opCtx) {
+    auto findRequest = FindCommandRequest(NamespaceString::kConfigsvrRestoreNamespace);
+
+    std::set<std::string> databasesToRestore;
+    DBDirectClient client(opCtx);
+    auto it = client.find(findRequest);
+    while (it->more()) {
+        const auto doc = it->next();
+        if (!doc.hasField("ns")) {
+            continue;
+        }
+
+        NamespaceString nss(doc.getStringField("ns"));
+        databasesToRestore.emplace(nss.db());
+    }
+
+    return databasesToRestore;
+}
+
 // Modifications to this map should add new testing in 'sharded_backup_restore.js'.
 // { config collection namespace -> ( optional nss field name, optional UUID field name ) }
 const stdx::unordered_map<NamespaceString,
@@ -146,7 +165,7 @@ public:
 
         // Keeps track of database names for collections restored. Databases with no collections
         // restored will have their entries removed in the config collections.
-        std::set<std::string> databasesRestored;
+        std::set<std::string> databasesRestored = getDatabasesToRestore(opCtx);
 
         for (const auto& collectionEntry : kCollectionEntries) {
             const NamespaceString& nss = collectionEntry.first;
@@ -200,10 +219,6 @@ public:
                             "doc"_attr = doc,
                             "shouldRestore"_attr = shouldRestore);
 
-                if (shouldRestore == ShouldRestoreDocument::kYes && docNss) {
-                    databasesRestored.insert(docNss->db().toString());
-                }
-
                 if (shouldRestore == ShouldRestoreDocument::kYes ||
                     shouldRestore == ShouldRestoreDocument::kMaybe) {
                     continue;
diff --git a/src/mongo/db/s/config/configsvr_set_cluster_parameter_command.cpp b/src/mongo/db/s/config/configsvr_set_cluster_parameter_command.cpp
index da76faa6593..24030e5fe57 100644
--- a/src/mongo/db/s/config/configsvr_set_cluster_parameter_command.cpp
+++ b/src/mongo/db/s/config/configsvr_set_cluster_parameter_command.cpp
@@ -61,6 +61,11 @@ public:
 
             const auto coordinatorCompletionFuture = [&]() -> SharedSemiFuture<void> {
                 FixedFCVRegion fcvRegion(opCtx);
+                uassert(ErrorCodes::UnknownFeatureCompatibilityVersion,
+                        "FCV is not yet initialized, retry the command after FCV initialization "
+                        "has completed",
+                        serverGlobalParams.featureCompatibility.isVersionInitialized());
+
                 uassert(ErrorCodes::IllegalOperation,
                         "featureFlagClusterWideConfig not enabled",
                         gFeatureFlagClusterWideConfig.isEnabled(
diff --git a/src/mongo/db/s/config/initial_split_policy.cpp b/src/mongo/db/s/config/initial_split_policy.cpp
index 5bdfb55d6c3..55f2d1c534e 100644
--- a/src/mongo/db/s/config/initial_split_policy.cpp
+++ b/src/mongo/db/s/config/initial_split_policy.cpp
@@ -37,12 +37,14 @@
 #include "mongo/db/bson/dotted_path_support.h"
 #include "mongo/db/catalog/collection_catalog.h"
 #include "mongo/db/curop.h"
+#include "mongo/db/pipeline/document_source.h"
 #include "mongo/db/pipeline/lite_parsed_pipeline.h"
 #include "mongo/db/pipeline/process_interface/shardsvr_process_interface.h"
 #include "mongo/db/pipeline/sharded_agg_helpers.h"
 #include "mongo/db/s/balancer/balancer_policy.h"
 #include "mongo/db/s/sharding_state.h"
 #include "mongo/db/vector_clock.h"
+#include "mongo/logv2/log.h"
 #include "mongo/s/balancer_configuration.h"
 #include "mongo/s/catalog/type_shard.h"
 #include "mongo/s/grid.h"
@@ -54,6 +56,7 @@ namespace {
 
 using ChunkDistributionMap = stdx::unordered_map<ShardId, size_t>;
 using ZoneShardMap = StringMap<std::vector<ShardId>>;
+using boost::intrusive_ptr;
 
 std::vector<ShardId> getAllShardIdsSorted(OperationContext* opCtx) {
     // Many tests assume that chunks will be placed on shards
@@ -266,7 +269,8 @@ std::unique_ptr<InitialSplitPolicy> InitialSplitPolicy::calculateOptimizationStr
     const boost::optional<std::vector<BSONObj>>& initialSplitPoints,
     const std::vector<TagsType>& tags,
     size_t numShards,
-    bool collectionIsEmpty) {
+    bool collectionIsEmpty,
+    bool useAutoSplitter) {
     uassert(ErrorCodes::InvalidOptions,
             str::stream() << "numInitialChunks is only supported when the collection is empty "
                              "and has a hashed field in the shard key pattern",
@@ -310,7 +314,11 @@ std::unique_ptr<InitialSplitPolicy> InitialSplitPolicy::calculateOptimizationStr
         return std::make_unique<SingleChunkOnPrimarySplitPolicy>();
     }
 
-    return std::make_unique<UnoptimizedSplitPolicy>();
+    if (useAutoSplitter) {
+        return std::make_unique<AutoSplitInChunksOnPrimaryPolicy>();
+    }
+
+    return std::make_unique<SingleChunkOnPrimarySplitPolicy>();
 }
 
 InitialSplitPolicy::ShardCollectionConfig SingleChunkOnPrimarySplitPolicy::createFirstChunks(
@@ -334,7 +342,7 @@ InitialSplitPolicy::ShardCollectionConfig SingleChunkOnPrimarySplitPolicy::creat
     return {std::move(chunks)};
 }
 
-InitialSplitPolicy::ShardCollectionConfig UnoptimizedSplitPolicy::createFirstChunks(
+InitialSplitPolicy::ShardCollectionConfig AutoSplitInChunksOnPrimaryPolicy::createFirstChunks(
     OperationContext* opCtx,
     const ShardKeyPattern& shardKeyPattern,
     const SplitPolicyParams& params) {
@@ -650,34 +658,29 @@ std::vector<BSONObj> ReshardingSplitPolicy::createRawPipeline(const ShardKeyPatt
 
     std::vector<BSONObj> res;
     const auto& shardKeyFields = shardKey.getKeyPatternFields();
-
-    BSONObjBuilder projectValBuilder;
     BSONObjBuilder sortValBuilder;
-
+    using Doc = Document;
+    using Arr = std::vector<Value>;
+    using V = Value;
+    Arr arrayToObjectBuilder;
     for (auto&& fieldRef : shardKeyFields) {
         // If the shard key includes a hashed field and current fieldRef is the hashed field.
         if (shardKey.isHashedPattern() &&
             fieldRef->dottedField().compare(shardKey.getHashedField().fieldNameStringData()) == 0) {
-            projectValBuilder.append(fieldRef->dottedField(),
-                                     BSON("$toHashedIndexKey"
-                                          << "$" + fieldRef->dottedField()));
+            arrayToObjectBuilder.emplace_back(
+                Doc{{"k", V{fieldRef->dottedField()}},
+                    {"v", Doc{{"$toHashedIndexKey", V{"$" + fieldRef->dottedField()}}}}});
         } else {
-            projectValBuilder.append(
-                str::stream() << fieldRef->dottedField(),
-                BSON("$ifNull" << BSON_ARRAY("$" + fieldRef->dottedField() << BSONNULL)));
+            arrayToObjectBuilder.emplace_back(Doc{
+                {"k", V{fieldRef->dottedField()}},
+                {"v", Doc{{"$ifNull", V{Arr{V{"$" + fieldRef->dottedField()}, V{BSONNULL}}}}}}});
         }
-
         sortValBuilder.append(fieldRef->dottedField().toString(), 1);
     }
-
-    // Do not project _id if it's not part of the shard key.
-    if (!shardKey.hasId()) {
-        projectValBuilder.append("_id", 0);
-    }
-
     res.push_back(BSON("$sample" << BSON("size" << numSplitPoints * samplesPerChunk)));
-    res.push_back(BSON("$project" << projectValBuilder.obj()));
     res.push_back(BSON("$sort" << sortValBuilder.obj()));
+    res.push_back(
+        Doc{{"$replaceWith", Doc{{"$arrayToObject", Arr{V{arrayToObjectBuilder}}}}}}.toBson());
     return res;
 }
 
@@ -800,26 +803,34 @@ void ReshardingSplitPolicy::_appendSplitPointsFromSample(BSONObjSet* splitPoints
 
     while (nextKey && nRemaining > 0) {
         // if key is hashed, nextKey values are already hashed
-        auto result = splitPoints->insert(
-            dotted_path_support::extractElementsBasedOnTemplate(*nextKey, shardKey.toBSON())
-                .getOwned());
-
+        auto result = splitPoints->insert(nextKey->getOwned());
         if (result.second) {
             nRemaining--;
         }
-
         nextKey = _samples->getNext();
     }
 }
 
 std::unique_ptr<ReshardingSplitPolicy::SampleDocumentSource>
+ReshardingSplitPolicy::makePipelineDocumentSource_forTest(OperationContext* opCtx,
+                                                          const NamespaceString& ns,
+                                                          const ShardKeyPattern& shardKey,
+                                                          int numInitialChunks,
+                                                          int samplesPerChunk) {
+    MakePipelineOptions opts;
+    opts.attachCursorSource = false;
+    return _makePipelineDocumentSource(
+        opCtx, ns, shardKey, numInitialChunks, samplesPerChunk, std::move(opts));
+}
+
+std::unique_ptr<ReshardingSplitPolicy::SampleDocumentSource>
 ReshardingSplitPolicy::_makePipelineDocumentSource(OperationContext* opCtx,
                                                    const NamespaceString& ns,
                                                    const ShardKeyPattern& shardKey,
                                                    int numInitialChunks,
-                                                   int samplesPerChunk) {
+                                                   int samplesPerChunk,
+                                                   MakePipelineOptions opts) {
     auto rawPipeline = createRawPipeline(shardKey, numInitialChunks - 1, samplesPerChunk);
-
     StringMap<ExpressionContext::ResolvedNamespace> resolvedNamespaces;
     resolvedNamespaces[ns.coll()] = {ns, std::vector<BSONObj>{}};
 
@@ -833,7 +844,7 @@ ReshardingSplitPolicy::_makePipelineDocumentSource(OperationContext* opCtx,
                                                     boost::none, /* explain */
                                                     false,       /* fromMongos */
                                                     false,       /* needsMerge */
-                                                    false,       /* allowDiskUse */
+                                                    true,        /* allowDiskUse */
                                                     true,        /* bypassDocumentValidation */
                                                     false,       /* isMapReduceCommand */
                                                     ns,
@@ -843,8 +854,10 @@ ReshardingSplitPolicy::_makePipelineDocumentSource(OperationContext* opCtx,
                                                     std::move(resolvedNamespaces),
                                                     boost::none); /* collUUID */
 
-    return std::make_unique<PipelineDocumentSource>(Pipeline::makePipeline(rawPipeline, expCtx, {}),
-                                                    samplesPerChunk - 1);
+    expCtx->tempDir = storageGlobalParams.dbpath + "/tmp";
+
+    return std::make_unique<PipelineDocumentSource>(
+        Pipeline::makePipeline(rawPipeline, expCtx, opts), samplesPerChunk - 1);
 }
 
 ReshardingSplitPolicy::PipelineDocumentSource::PipelineDocumentSource(
diff --git a/src/mongo/db/s/config/initial_split_policy.h b/src/mongo/db/s/config/initial_split_policy.h
index ced8d519a13..e492e9c4cb2 100644
--- a/src/mongo/db/s/config/initial_split_policy.h
+++ b/src/mongo/db/s/config/initial_split_policy.h
@@ -40,7 +40,6 @@
 #include "mongo/s/shard_id.h"
 #include "mongo/s/shard_key_pattern.h"
 #include "mongo/util/string_map.h"
-
 namespace mongo {
 
 struct SplitPolicyParams {
@@ -53,6 +52,11 @@ public:
     /**
      * Returns the optimization strategy for building initial chunks based on the input parameters
      * and the collection state.
+     *
+     * The 'useAutoSplitter' flag indicates to the initial split strategy selected that in case the
+     * collection contains data, it should use the auto splitter to chop that data into chunks
+     * respective to the configured chunk size. If set to false, the policy will create as large of
+     * chunks as possible.
      */
     static std::unique_ptr<InitialSplitPolicy> calculateOptimizationStrategy(
         OperationContext* opCtx,
@@ -62,7 +66,8 @@ public:
         const boost::optional<std::vector<BSONObj>>& initialSplitPoints,
         const std::vector<TagsType>& tags,
         size_t numShards,
-        bool collectionIsEmpty);
+        bool collectionIsEmpty,
+        bool useAutoSplitter = true /* Controlled by FCV, see the comment */);
 
     virtual ~InitialSplitPolicy() {}
 
@@ -140,7 +145,7 @@ public:
  * Split point building strategy to be used when no optimizations are available. We send a
  * splitVector command to the primary shard in order to calculate the appropriate split points.
  */
-class UnoptimizedSplitPolicy : public InitialSplitPolicy {
+class AutoSplitInChunksOnPrimaryPolicy : public InitialSplitPolicy {
 public:
     ShardCollectionConfig createFirstChunks(OperationContext* opCtx,
                                             const ShardKeyPattern& shardKeyPattern,
@@ -285,6 +290,7 @@ public:
     public:
         virtual ~SampleDocumentSource(){};
         virtual boost::optional<BSONObj> getNext() = 0;
+        virtual Pipeline* getPipeline_forTest() = 0;
     };
 
     // Provides documents from a real Pipeline
@@ -293,6 +299,9 @@ public:
         PipelineDocumentSource() = delete;
         PipelineDocumentSource(SampleDocumentPipeline pipeline, int skip);
         boost::optional<BSONObj> getNext() override;
+        Pipeline* getPipeline_forTest() override {
+            return _pipeline.get();
+        }
 
     private:
         SampleDocumentPipeline _pipeline;
@@ -328,13 +337,21 @@ public:
 
     static constexpr int kDefaultSamplesPerChunk = 10;
 
+    static std::unique_ptr<SampleDocumentSource> makePipelineDocumentSource_forTest(
+        OperationContext* opCtx,
+        const NamespaceString& ns,
+        const ShardKeyPattern& shardKey,
+        int numInitialChunks,
+        int samplesPerChunk);
+
 private:
     static std::unique_ptr<SampleDocumentSource> _makePipelineDocumentSource(
         OperationContext* opCtx,
         const NamespaceString& ns,
         const ShardKeyPattern& shardKey,
         int numInitialChunks,
-        int samplesPerChunk);
+        int samplesPerChunk,
+        MakePipelineOptions opts = {});
 
     /**
      * Returns a set of split points to ensure that chunk boundaries will align with the zone
diff --git a/src/mongo/db/s/config/initial_split_policy_test.cpp b/src/mongo/db/s/config/initial_split_policy_test.cpp
index 75d1f2ae0a1..0ef3bee06d1 100644
--- a/src/mongo/db/s/config/initial_split_policy_test.cpp
+++ b/src/mongo/db/s/config/initial_split_policy_test.cpp
@@ -35,6 +35,7 @@
 #include "mongo/db/s/config/config_server_test_fixture.h"
 #include "mongo/db/s/config/initial_split_policy.h"
 #include "mongo/db/vector_clock.h"
+#include "mongo/logv2/log.h"
 #include "mongo/s/catalog/type_shard.h"
 #include "mongo/s/catalog/type_tags.h"
 #include "mongo/unittest/unittest.h"
@@ -1727,6 +1728,10 @@ public:
         return next;
     }
 
+    Pipeline* getPipeline_forTest() override {
+        return nullptr;
+    }
+
 private:
     std::list<BSONObj> _toReturn;
 };
@@ -1765,9 +1770,9 @@ TEST_F(ReshardingInitSplitTest, NoZones) {
     shardRegistry()->reload(operationContext());
 
     std::list<BSONObj> mockSamples;
-    mockSamples.push_back(BSON("x" << 10 << "y" << 10));
-    mockSamples.push_back(BSON("x" << 10 << "y" << 20));
-    mockSamples.push_back(BSON("x" << 10 << "y" << 30));
+    mockSamples.push_back(BSON("y" << 10));
+    mockSamples.push_back(BSON("y" << 20));
+    mockSamples.push_back(BSON("y" << 30));
 
     auto mockSampleSource = std::make_unique<MockPipelineSource>(std::move(mockSamples));
 
@@ -1802,9 +1807,9 @@ TEST_F(ReshardingInitSplitTest, HashedShardKey) {
     shardRegistry()->reload(operationContext());
 
     std::list<BSONObj> mockSamples;
-    mockSamples.push_back(BSON("x" << 10 << "y" << 7766103514953448109LL));
-    mockSamples.push_back(BSON("x" << 10 << "y" << -9117533237618642180LL));
-    mockSamples.push_back(BSON("x" << 10 << "y" << -1196399207910989725LL));
+    mockSamples.push_back(BSON("y" << 7766103514953448109LL));
+    mockSamples.push_back(BSON("y" << -9117533237618642180LL));
+    mockSamples.push_back(BSON("y" << -1196399207910989725LL));
 
     auto mockSampleSource = std::make_unique<MockPipelineSource>(std::move(mockSamples));
 
@@ -1867,9 +1872,9 @@ TEST_F(ReshardingInitSplitTest, ZonesCoversEntireDomainButInsufficient) {
     shardRegistry()->reload(operationContext());
 
     std::list<BSONObj> mockSamples;
-    mockSamples.push_back(BSON("x" << 10 << "y" << 10));
-    mockSamples.push_back(BSON("x" << 10 << "y" << 20));
-    mockSamples.push_back(BSON("x" << 10 << "y" << 30));
+    mockSamples.push_back(BSON("y" << 10));
+    mockSamples.push_back(BSON("y" << 20));
+    mockSamples.push_back(BSON("y" << 30));
 
     auto mockSampleSource = std::make_unique<MockPipelineSource>(std::move(mockSamples));
 
@@ -1907,9 +1912,9 @@ TEST_F(ReshardingInitSplitTest, SamplesCoincidingWithZones) {
     shardRegistry()->reload(operationContext());
 
     std::list<BSONObj> mockSamples;
-    mockSamples.push_back(BSON("x" << 10 << "y" << 10));
-    mockSamples.push_back(BSON("x" << 10 << "y" << 20));
-    mockSamples.push_back(BSON("x" << 10 << "y" << 30));
+    mockSamples.push_back(BSON("y" << 10));
+    mockSamples.push_back(BSON("y" << 20));
+    mockSamples.push_back(BSON("y" << 30));
 
     auto mockSampleSource = std::make_unique<MockPipelineSource>(std::move(mockSamples));
 
diff --git a/src/mongo/db/s/config/sharding_catalog_manager.h b/src/mongo/db/s/config/sharding_catalog_manager.h
index 070c1d3d78f..8ec8fb02108 100644
--- a/src/mongo/db/s/config/sharding_catalog_manager.h
+++ b/src/mongo/db/s/config/sharding_catalog_manager.h
@@ -31,6 +31,7 @@
 
 #include "mongo/base/status_with.h"
 #include "mongo/bson/bsonobj.h"
+#include "mongo/client/fetcher.h"
 #include "mongo/db/auth/authorization_session.h"
 #include "mongo/db/concurrency/d_concurrency.h"
 #include "mongo/db/logical_session_cache.h"
@@ -271,8 +272,7 @@ public:
                                           const boost::optional<Timestamp>& timestamp,
                                           const UUID& requestCollectionUUID,
                                           const ChunkRange& chunkRange,
-                                          const ShardId& shardId,
-                                          const boost::optional<Timestamp>& validAfter);
+                                          const ShardId& shardId);
 
     /**
      * Updates metadata in config.chunks collection to show the given chunk in its new shard.
@@ -348,7 +348,8 @@ public:
      */
     void splitOrMarkJumbo(OperationContext* opCtx,
                           const NamespaceString& nss,
-                          const BSONObj& minKey);
+                          const BSONObj& minKey,
+                          boost::optional<int64_t> optMaxChunkSizeBytes);
 
     /**
      * In a transaction, sets the 'allowMigrations' to the requested state and bumps the collection
@@ -396,6 +397,15 @@ public:
         // # TODO SERVER-63983: remove enableSharding paramter when 6.0 becomes lastLTS
         bool enableSharding = false);
 
+    /**
+     * Updates the metadata in config.databases collection with the new primary shard for the given
+     * database. This also advances the database's lastmod.
+     */
+    void commitMovePrimary(OperationContext* opCtx,
+                           const StringData& dbName,
+                           const DatabaseVersion& expectedDbVersion,
+                           const ShardId& toShardId);
+
     //
     // Collection Operations
     //
@@ -596,10 +606,11 @@ private:
      * Runs a command against a "shard" that is not yet in the cluster and thus not present in the
      * ShardRegistry.
      */
-    StatusWith<Shard::CommandResponse> _runCommandForAddShard(OperationContext* opCtx,
-                                                              RemoteCommandTargeter* targeter,
-                                                              StringData dbName,
-                                                              const BSONObj& cmdObj);
+    StatusWith<Shard::CommandResponse> _runCommandForAddShard(
+        OperationContext* opCtx,
+        std::shared_ptr<RemoteCommandTargeter> targeter,
+        StringData dbName,
+        const BSONObj& cmdObj);
 
     /**
      * Helper method for running a count command against the config server with appropriate error
@@ -636,7 +647,29 @@ private:
      * Sets the current cluster's user-write blocking state on the shard that is being added.
      */
     void _setUserWriteBlockingStateOnNewShard(OperationContext* opCtx,
-                                              RemoteCommandTargeter* targeter);
+                                              std::shared_ptr<RemoteCommandTargeter> targeter);
+
+    using FetcherDocsCallbackFn = std::function<bool(const std::vector<BSONObj>& batch)>;
+    using FetcherStatusCallbackFn = std::function<void(const Status& status)>;
+
+    /**
+     * Creates a Fetcher task for fetching documents in the given collection on the given shard.
+     * After the task is scheduled, applies 'processDocsCallback' to each fetched batch and
+     * 'processStatusCallback' to the fetch status.
+     */
+    std::unique_ptr<Fetcher> _createFetcher(OperationContext* opCtx,
+                                            std::shared_ptr<RemoteCommandTargeter> targeter,
+                                            const NamespaceString& nss,
+                                            const repl::ReadConcernLevel& readConcernLevel,
+                                            FetcherDocsCallbackFn processDocsCallback,
+                                            FetcherStatusCallbackFn processStatusCallback);
+
+    /**
+     * Gets the cluster time keys on the given shard and then saves them locally.
+     */
+    Status _pullClusterTimeKeys(OperationContext* opCtx,
+                                std::shared_ptr<RemoteCommandTargeter> targeter);
+
     /**
      * Given a vector of cluster parameters in disk format, sets them locally.
      */
@@ -647,14 +680,14 @@ private:
      * Gets the cluster parameters set on the shard and then saves them locally.
      */
     void _pullClusterParametersFromNewShard(OperationContext* opCtx,
-                                            RemoteCommandTargeter* targeter);
+                                            std::shared_ptr<RemoteCommandTargeter> targeter);
 
     /**
      * Clean all possible leftover cluster parameters on the new added shard and sets the ones
      * stored on the config server.
      */
     void _pushClusterParametersToNewShard(OperationContext* opCtx,
-                                          RemoteCommandTargeter* targeter,
+                                          std::shared_ptr<RemoteCommandTargeter> targeter,
                                           const std::vector<BSONObj>& clusterParameters);
 
     /**
@@ -662,7 +695,20 @@ private:
      * converting from a replica set to a sharded cluster) or set the cluster parameters stored on
      * the config server in the newly added shard.
      */
-    void _standardizeClusterParameters(OperationContext* opCtx, RemoteCommandTargeter* targeter);
+    void _standardizeClusterParameters(OperationContext* opCtx,
+                                       std::shared_ptr<RemoteCommandTargeter> targeter);
+
+    /**
+     * Execute the merge chunk updates using the internal transaction API.
+     */
+    void _mergeChunksInTransaction(OperationContext* opCtx,
+                                   const NamespaceString& nss,
+                                   const UUID& collectionUUID,
+                                   const ChunkVersion& mergeVersion,
+                                   const Timestamp& validAfter,
+                                   const ChunkRange& chunkRange,
+                                   const ShardId& shardId,
+                                   std::shared_ptr<std::vector<ChunkType>> chunksToMerge);
 
     // The owning service context
     ServiceContext* const _serviceContext;
diff --git a/src/mongo/db/s/config/sharding_catalog_manager_add_shard_test.cpp b/src/mongo/db/s/config/sharding_catalog_manager_add_shard_test.cpp
index 0ed9c76d0f0..db1d21033bd 100644
--- a/src/mongo/db/s/config/sharding_catalog_manager_add_shard_test.cpp
+++ b/src/mongo/db/s/config/sharding_catalog_manager_add_shard_test.cpp
@@ -41,11 +41,13 @@
 #include "mongo/db/ops/write_ops.h"
 #include "mongo/db/query/cursor_response.h"
 #include "mongo/db/repl/replication_coordinator_mock.h"
+#include "mongo/db/repl/wait_for_majority_service.h"
 #include "mongo/db/s/add_shard_cmd_gen.h"
 #include "mongo/db/s/add_shard_util.h"
 #include "mongo/db/s/config/config_server_test_fixture.h"
 #include "mongo/db/s/config/sharding_catalog_manager.h"
 #include "mongo/db/s/type_shard_identity.h"
+#include "mongo/db/time_proof_service.h"
 #include "mongo/idl/cluster_server_parameter_gen.h"
 #include "mongo/s/catalog/config_server_version.h"
 #include "mongo/s/catalog/type_changelog.h"
@@ -83,25 +85,32 @@ protected:
         ASSERT_OK(clusterIdLoader->loadClusterId(operationContext(),
                                                  repl::ReadConcernLevel::kLocalReadConcern));
         _clusterId = clusterIdLoader->getClusterId();
+
+        WaitForMajorityService::get(getServiceContext()).startup(getServiceContext());
+    }
+
+    void tearDown() override {
+        WaitForMajorityService::get(getServiceContext()).shutDown();
+        ConfigServerTestFixture::tearDown();
     }
 
     /**
-     * addShard validates the host as a shard. It calls "isMaster" on the host to determine what
+     * addShard validates the host as a shard. It calls "hello" on the host to determine what
      * kind of host it is -- mongos, regular mongod, config mongod -- and whether the replica set
-     * details are correct. "isMasterResponse" defines the response of the "isMaster" request and
+     * details are correct. "helloResponse" defines the response of the "hello" request and
      * should be a command response BSONObj, or a failed Status.
      *
      * ShardingTestFixture::expectGetShards() should be called before this function, otherwise
-     * addShard will never reach the isMaster command -- a find query is called first.
+     * addShard will never reach the "hello" command -- a find query is called first.
      */
-    void expectIsMaster(const HostAndPort& target, StatusWith<BSONObj> isMasterResponse) {
-        onCommandForAddShard([&, target, isMasterResponse](const RemoteCommandRequest& request) {
+    void expectHello(const HostAndPort& target, StatusWith<BSONObj> helloResponse) {
+        onCommandForAddShard([&, target, helloResponse](const RemoteCommandRequest& request) {
             ASSERT_EQ(request.target, target);
             ASSERT_EQ(request.dbname, "admin");
-            ASSERT_BSONOBJ_EQ(request.cmdObj, BSON("isMaster" << 1));
+            ASSERT_BSONOBJ_EQ(request.cmdObj, BSON("hello" << 1));
             ASSERT_BSONOBJ_EQ(rpc::makeEmptyMetadata(), request.metadata);
 
-            return isMasterResponse;
+            return helloResponse;
         });
     }
 
@@ -225,7 +234,7 @@ protected:
             ASSERT_EQ(request.dbname, NamespaceString::kClusterParametersNamespace.db());
             ASSERT_BSONOBJ_EQ(request.cmdObj,
                               BSON("find" << NamespaceString::kClusterParametersNamespace.coll()
-                                          << "maxTimeMS" << 30000 << "readConcern"
+                                          << "maxTimeMS" << 60000 << "readConcern"
                                           << BSON("level"
                                                   << "majority")));
             auto cursorRes = CursorResponse(NamespaceString::kClusterParametersNamespace, 0, {});
@@ -233,6 +242,24 @@ protected:
         });
     }
 
+    void expectClusterTimeKeysPullRequest(const HostAndPort& target) {
+        onCommandForAddShard([&](const RemoteCommandRequest& request) {
+            ASSERT_EQ(request.target, target);
+            ASSERT_BSONOBJ_EQ(request.cmdObj,
+                              BSON("find" << NamespaceString::kKeysCollectionNamespace.coll()
+                                          << "maxTimeMS" << 60000 << "readConcern"
+                                          << BSON("level"
+                                                  << "local")));
+
+            KeysCollectionDocument key(1);
+            key.setKeysCollectionDocumentBase(
+                {"dummy", TimeProofService::generateRandomKey(), LogicalTime(Timestamp(105, 0))});
+            auto cursorRes =
+                CursorResponse(NamespaceString::kKeysCollectionNamespace, 0, {key.toBSON()});
+            return cursorRes.toBSON(CursorResponse::ResponseType::InitialResponse);
+        });
+    }
+
     /**
      * Waits for a request for the shardIdentity document to be upserted into a shard from the
      * config server on addShard.
@@ -482,9 +509,9 @@ TEST_F(AddShardTest, StandaloneBasicSuccess) {
         ASSERT_EQUALS(expectedShardName, shardName);
     });
 
-    BSONObj commandResponse = BSON("ok" << 1 << "ismaster" << true << "maxWireVersion"
+    BSONObj commandResponse = BSON("ok" << 1 << "isWritablePrimary" << true << "maxWireVersion"
                                         << WireVersion::LATEST_WIRE_VERSION);
-    expectIsMaster(shardTarget, commandResponse);
+    expectHello(shardTarget, commandResponse);
 
     // Get databases list from new shard
     expectListDatabases(
@@ -497,6 +524,9 @@ TEST_F(AddShardTest, StandaloneBasicSuccess) {
 
     expectCollectionDrop(shardTarget, NamespaceString("config", "system.sessions"));
 
+    // The shard receives a find to pull all clusterTime keys from the new shard.
+    expectClusterTimeKeysPullRequest(shardTarget);
+
     // The shard receives the _addShard command
     expectAddShardCmdReturnSuccess(shardTarget, expectedShardName);
 
@@ -569,9 +599,9 @@ TEST_F(AddShardTest, StandaloneGenerateName) {
         ASSERT_EQUALS(expectedShardName, shardName);
     });
 
-    BSONObj commandResponse = BSON("ok" << 1 << "ismaster" << true << "maxWireVersion"
+    BSONObj commandResponse = BSON("ok" << 1 << "isWritablePrimary" << true << "maxWireVersion"
                                         << WireVersion::LATEST_WIRE_VERSION);
-    expectIsMaster(shardTarget, commandResponse);
+    expectHello(shardTarget, commandResponse);
 
     // Get databases list from new shard
     expectListDatabases(
@@ -584,6 +614,9 @@ TEST_F(AddShardTest, StandaloneGenerateName) {
 
     expectCollectionDrop(shardTarget, NamespaceString("config", "system.sessions"));
 
+    // The shard receives a find to pull all clusterTime keys from the new shard.
+    expectClusterTimeKeysPullRequest(shardTarget);
+
     // The shard receives the _addShard command
     expectAddShardCmdReturnSuccess(shardTarget, expectedShardName);
 
@@ -673,7 +706,7 @@ TEST_F(AddShardTest, UnreachableHost) {
     });
 
     Status hostUnreachableStatus = Status(ErrorCodes::HostUnreachable, "host unreachable");
-    expectIsMaster(shardTarget, hostUnreachableStatus);
+    expectHello(shardTarget, hostUnreachableStatus);
 
     future.timed_get(kLongFutureTimeout);
 }
@@ -698,9 +731,9 @@ TEST_F(AddShardTest, AddMongosAsShard) {
         ASSERT_EQUALS(ErrorCodes::IllegalOperation, status);
     });
 
-    expectIsMaster(shardTarget,
-                   BSON("msg"
-                        << "isdbgrid"));
+    expectHello(shardTarget,
+                BSON("msg"
+                     << "isdbgrid"));
 
     future.timed_get(kLongFutureTimeout);
 }
@@ -726,10 +759,10 @@ TEST_F(AddShardTest, AddReplicaSetShardAsStandalone) {
         ASSERT_STRING_CONTAINS(status.getStatus().reason(), "use replica set url format");
     });
 
-    BSONObj commandResponse = BSON("ok" << 1 << "ismaster" << true << "setName"
+    BSONObj commandResponse = BSON("ok" << 1 << "isWritablePrimary" << true << "setName"
                                         << "myOtherSet"
                                         << "maxWireVersion" << WireVersion::LATEST_WIRE_VERSION);
-    expectIsMaster(shardTarget, commandResponse);
+    expectHello(shardTarget, commandResponse);
 
     future.timed_get(kLongFutureTimeout);
 }
@@ -756,9 +789,9 @@ TEST_F(AddShardTest, AddStandaloneHostShardAsReplicaSet) {
         ASSERT_STRING_CONTAINS(status.getStatus().reason(), "host did not return a set name");
     });
 
-    BSONObj commandResponse = BSON("ok" << 1 << "ismaster" << true << "maxWireVersion"
+    BSONObj commandResponse = BSON("ok" << 1 << "isWritablePrimary" << true << "maxWireVersion"
                                         << WireVersion::LATEST_WIRE_VERSION);
-    expectIsMaster(shardTarget, commandResponse);
+    expectHello(shardTarget, commandResponse);
 
     future.timed_get(kLongFutureTimeout);
 }
@@ -785,10 +818,10 @@ TEST_F(AddShardTest, ReplicaSetMistmatchedReplicaSetName) {
         ASSERT_STRING_CONTAINS(status.getStatus().reason(), "does not match the actual set name");
     });
 
-    BSONObj commandResponse = BSON("ok" << 1 << "ismaster" << true << "setName"
+    BSONObj commandResponse = BSON("ok" << 1 << "isWritablePrimary" << true << "setName"
                                         << "myOtherSet"
                                         << "maxWireVersion" << WireVersion::LATEST_WIRE_VERSION);
-    expectIsMaster(shardTarget, commandResponse);
+    expectHello(shardTarget, commandResponse);
 
     future.timed_get(kLongFutureTimeout);
 }
@@ -817,10 +850,10 @@ TEST_F(AddShardTest, ShardIsCSRSConfigServer) {
     });
 
     BSONObj commandResponse =
-        BSON("ok" << 1 << "ismaster" << true << "setName"
+        BSON("ok" << 1 << "isWritablePrimary" << true << "setName"
                   << "config"
                   << "configsvr" << true << "maxWireVersion" << WireVersion::LATEST_WIRE_VERSION);
-    expectIsMaster(shardTarget, commandResponse);
+    expectHello(shardTarget, commandResponse);
 
     future.timed_get(kLongFutureTimeout);
 }
@@ -850,11 +883,11 @@ TEST_F(AddShardTest, ReplicaSetMissingHostsProvidedInSeedList) {
 
     BSONArrayBuilder hosts;
     hosts.append("host1:12345");
-    BSONObj commandResponse = BSON("ok" << 1 << "ismaster" << true << "setName"
+    BSONObj commandResponse = BSON("ok" << 1 << "isWritablePrimary" << true << "setName"
                                         << "mySet"
                                         << "hosts" << hosts.arr() << "maxWireVersion"
                                         << WireVersion::LATEST_WIRE_VERSION);
-    expectIsMaster(shardTarget, commandResponse);
+    expectHello(shardTarget, commandResponse);
 
     future.timed_get(kLongFutureTimeout);
 }
@@ -885,11 +918,11 @@ TEST_F(AddShardTest, AddShardWithNameConfigFails) {
     BSONArrayBuilder hosts;
     hosts.append("host1:12345");
     hosts.append("host2:12345");
-    BSONObj commandResponse = BSON("ok" << 1 << "ismaster" << true << "setName"
+    BSONObj commandResponse = BSON("ok" << 1 << "isWritablePrimary" << true << "setName"
                                         << "mySet"
                                         << "hosts" << hosts.arr() << "maxWireVersion"
                                         << WireVersion::LATEST_WIRE_VERSION);
-    expectIsMaster(shardTarget, commandResponse);
+    expectHello(shardTarget, commandResponse);
 
     future.timed_get(kLongFutureTimeout);
 }
@@ -931,11 +964,11 @@ TEST_F(AddShardTest, ShardContainsExistingDatabase) {
     BSONArrayBuilder hosts;
     hosts.append("host1:12345");
     hosts.append("host2:12345");
-    BSONObj commandResponse = BSON("ok" << 1 << "ismaster" << true << "setName"
+    BSONObj commandResponse = BSON("ok" << 1 << "isWritablePrimary" << true << "setName"
                                         << "mySet"
                                         << "hosts" << hosts.arr() << "maxWireVersion"
                                         << WireVersion::LATEST_WIRE_VERSION);
-    expectIsMaster(shardTarget, commandResponse);
+    expectHello(shardTarget, commandResponse);
 
     expectListDatabases(shardTarget, {BSON("name" << existingDB.getName())});
 
@@ -975,17 +1008,20 @@ TEST_F(AddShardTest, SuccessfullyAddReplicaSet) {
     BSONArrayBuilder hosts;
     hosts.append("host1:12345");
     hosts.append("host2:12345");
-    BSONObj commandResponse = BSON("ok" << 1 << "ismaster" << true << "setName"
+    BSONObj commandResponse = BSON("ok" << 1 << "isWritablePrimary" << true << "setName"
                                         << "mySet"
                                         << "hosts" << hosts.arr() << "maxWireVersion"
                                         << WireVersion::LATEST_WIRE_VERSION);
-    expectIsMaster(shardTarget, commandResponse);
+    expectHello(shardTarget, commandResponse);
 
     // Get databases list from new shard
     expectListDatabases(shardTarget, std::vector<BSONObj>{BSON("name" << discoveredDB.getName())});
 
     expectCollectionDrop(shardTarget, NamespaceString("config", "system.sessions"));
 
+    // The shard receives a find to pull all clusterTime keys from the new shard.
+    expectClusterTimeKeysPullRequest(shardTarget);
+
     // The shard receives the _addShard command
     expectAddShardCmdReturnSuccess(shardTarget, expectedShardName);
 
@@ -1047,17 +1083,20 @@ TEST_F(AddShardTest, ReplicaSetExtraHostsDiscovered) {
     BSONArrayBuilder hosts;
     hosts.append("host1:12345");
     hosts.append("host2:12345");
-    BSONObj commandResponse = BSON("ok" << 1 << "ismaster" << true << "setName"
+    BSONObj commandResponse = BSON("ok" << 1 << "isWritablePrimary" << true << "setName"
                                         << "mySet"
                                         << "hosts" << hosts.arr() << "maxWireVersion"
                                         << WireVersion::LATEST_WIRE_VERSION);
-    expectIsMaster(shardTarget, commandResponse);
+    expectHello(shardTarget, commandResponse);
 
     // Get databases list from new shard
     expectListDatabases(shardTarget, std::vector<BSONObj>{BSON("name" << discoveredDB.getName())});
 
     expectCollectionDrop(shardTarget, NamespaceString("config", "system.sessions"));
 
+    // The shard receives a find to pull all clusterTime keys from the new shard.
+    expectClusterTimeKeysPullRequest(shardTarget);
+
     // The shard receives the _addShard command
     expectAddShardCmdReturnSuccess(shardTarget, expectedShardName);
 
@@ -1128,9 +1167,9 @@ TEST_F(AddShardTest, AddShardSucceedsEvenIfAddingDBsFromNewShardFails) {
         ASSERT_EQUALS(expectedShardName, shardName);
     });
 
-    BSONObj commandResponse = BSON("ok" << 1 << "ismaster" << true << "maxWireVersion"
+    BSONObj commandResponse = BSON("ok" << 1 << "isWritablePrimary" << true << "maxWireVersion"
                                         << WireVersion::LATEST_WIRE_VERSION);
-    expectIsMaster(shardTarget, commandResponse);
+    expectHello(shardTarget, commandResponse);
 
     // Get databases list from new shard
     expectListDatabases(
@@ -1143,6 +1182,9 @@ TEST_F(AddShardTest, AddShardSucceedsEvenIfAddingDBsFromNewShardFails) {
 
     expectCollectionDrop(shardTarget, NamespaceString("config", "system.sessions"));
 
+    // The shard receives a find to pull all clusterTime keys from the new shard.
+    expectClusterTimeKeysPullRequest(shardTarget);
+
     // The shard receives the _addShard command
     expectAddShardCmdReturnSuccess(shardTarget, expectedShardName);
 
diff --git a/src/mongo/db/s/config/sharding_catalog_manager_chunk_operations.cpp b/src/mongo/db/s/config/sharding_catalog_manager_chunk_operations.cpp
index 172729b575b..dfdc04f8517 100644
--- a/src/mongo/db/s/config/sharding_catalog_manager_chunk_operations.cpp
+++ b/src/mongo/db/s/config/sharding_catalog_manager_chunk_operations.cpp
@@ -89,76 +89,6 @@ void appendShortVersion(BufBuilder* out, const ChunkType& chunk) {
     bb.done();
 }
 
-BSONArray buildMergeChunksTransactionUpdates(const std::vector<ChunkType>& chunksToMerge,
-                                             const ChunkVersion& mergeVersion,
-                                             const boost::optional<Timestamp>& validAfter) {
-    BSONArrayBuilder updates;
-
-    // Build an update operation to expand the first chunk into the newly merged chunk
-    {
-        BSONObjBuilder op;
-        op.append("op", "u");
-        op.appendBool("b", false);  // no upsert
-        op.append("ns", ChunkType::ConfigNS.ns());
-
-        // expand first chunk into newly merged chunk
-        ChunkType mergedChunk(chunksToMerge.front());
-        mergedChunk.setMax(chunksToMerge.back().getMax());
-
-        // fill in additional details for sending through transaction
-        mergedChunk.setVersion(mergeVersion);
-        mergedChunk.setEstimatedSizeBytes(boost::none);
-
-        invariant(validAfter);
-        mergedChunk.setHistory({ChunkHistory(validAfter.get(), mergedChunk.getShard())});
-
-        // add the new chunk information as the update object
-        op.append("o", mergedChunk.toConfigBSON());
-
-        // query object
-        op.append("o2", BSON(ChunkType::name(mergedChunk.getName())));
-
-        updates.append(op.obj());
-    }
-
-    // Build update operations to delete the rest of the chunks to be merged. Remember not
-    // to delete the first chunk we're expanding
-    for (size_t i = 1; i < chunksToMerge.size(); ++i) {
-        BSONObjBuilder op;
-        op.append("op", "d");
-        op.append("ns", ChunkType::ConfigNS.ns());
-
-        op.append("o", BSON(ChunkType::name(chunksToMerge[i].getName())));
-
-        updates.append(op.obj());
-    }
-
-    return updates.arr();
-}
-
-BSONArray buildMergeChunksTransactionPrecond(const std::vector<ChunkType>& chunksToMerge,
-                                             const ChunkVersion& collVersion) {
-    BSONArrayBuilder preCond;
-
-    for (const auto& chunk : chunksToMerge) {
-        BSONObj query = BSON(ChunkType::min(chunk.getMin())
-                             << ChunkType::max(chunk.getMax()) << ChunkType::collectionUUID()
-                             << chunk.getCollectionUUID());
-
-        const auto collectionIdentityMatchCondition =
-            BSON(ChunkType::collectionUUID()
-                 << chunk.getCollectionUUID() << ChunkType::shard(chunk.getShard().toString()));
-
-        BSONObjBuilder b;
-        b.append("ns", ChunkType::ConfigNS.ns());
-        b.append("q", BSON("query" << query << "orderby" << BSON(ChunkType::lastmod() << -1)));
-        b.append("res", collectionIdentityMatchCondition);
-
-        preCond.append(b.obj());
-    }
-    return preCond.arr();
-}
-
 /**
  * Check that the chunk still exists and return its metadata.
  */
@@ -166,13 +96,11 @@ StatusWith<ChunkType> findChunkContainingRange(OperationContext* opCtx,
                                                const UUID& uuid,
                                                const OID& epoch,
                                                const Timestamp& timestamp,
-                                               const BSONObj& min,
-                                               const BSONObj& max) {
+                                               const ChunkRange& range) {
     const auto chunkQuery = [&]() {
         BSONObjBuilder queryBuilder;
         queryBuilder << ChunkType::collectionUUID << uuid;
-        queryBuilder << ChunkType::min(BSON("$lte" << min));
-        queryBuilder << ChunkType::max(BSON("$gte" << max));
+        queryBuilder << ChunkType::min(BSON("$lte" << range.getMin()));
         return queryBuilder.obj();
     }();
 
@@ -184,21 +112,26 @@ StatusWith<ChunkType> findChunkContainingRange(OperationContext* opCtx,
             repl::ReadConcernLevel::kLocalReadConcern,
             ChunkType::ConfigNS,
             chunkQuery,
-            BSONObj(),
-            2 /* limit */);
+            BSON(ChunkType::min << -1),
+            1 /* limit */);
 
     if (!findResponseWith.isOK()) {
         return findResponseWith.getStatus();
     }
 
-    if (findResponseWith.getValue().docs.size() != 1) {
-        return {ErrorCodes::Error(40165),
-                str::stream() << "Could not find a chunk including bounds [" << min << ", " << max
-                              << "). Cannot execute the migration commit with invalid chunks."};
+    if (!findResponseWith.getValue().docs.empty()) {
+        const auto containingChunk = uassertStatusOK(ChunkType::parseFromConfigBSON(
+            findResponseWith.getValue().docs.front(), epoch, timestamp));
+
+        if (containingChunk.getRange().covers(range)) {
+            return containingChunk;
+        }
     }
 
-    return uassertStatusOK(
-        ChunkType::parseFromConfigBSON(findResponseWith.getValue().docs.front(), epoch, timestamp));
+    return {ErrorCodes::Error(40165),
+            str::stream() << "Could not find a chunk including bounds [" << range.getMin() << ", "
+                          << range.getMax()
+                          << "). Cannot execute the migration commit with invalid chunks."};
 }
 
 BSONObj makeCommitChunkTransactionCommand(const NamespaceString& nss,
@@ -570,6 +503,11 @@ StatusWith<BSONObj> ShardingCatalogManager::commitChunkSplit(
     const std::vector<BSONObj>& splitPoints,
     const std::string& shardName,
     const bool fromChunkSplitter) {
+
+    // Mark opCtx as interruptible to ensure that all reads and writes to the metadata collections
+    // under the exclusive _kChunkOpLock happen on the same term.
+    opCtx->setAlwaysInterruptAtStepDownOrUp();
+
     // Take _kChunkOpLock in exclusive mode to prevent concurrent chunk modifications and generate
     // strictly monotonously increasing collection versions
     Lock::ExclusiveLock lk(opCtx, opCtx->lockState(), _kChunkOpLock);
@@ -690,6 +628,7 @@ StatusWith<BSONObj> ShardingCatalogManager::commitChunkSplit(
         newChunk.setMin(startKey);
         newChunk.setMax(endKey);
         newChunk.setEstimatedSizeBytes(boost::none);
+        newChunk.setJumbo(false);
 
         op.append("o", newChunk.toConfigBSON());
 
@@ -784,6 +723,82 @@ StatusWith<BSONObj> ShardingCatalogManager::commitChunkSplit(
     return response.obj();
 }
 
+void ShardingCatalogManager::_mergeChunksInTransaction(
+    OperationContext* opCtx,
+    const NamespaceString& nss,
+    const UUID& collectionUUID,
+    const ChunkVersion& mergeVersion,
+    const Timestamp& validAfter,
+    const ChunkRange& chunkRange,
+    const ShardId& shardId,
+    std::shared_ptr<std::vector<ChunkType>> chunksToMerge) {
+    withTransaction(
+        opCtx, ChunkType::ConfigNS, [&, this](OperationContext* opCtx, TxnNumber txnNumber) {
+            // Construct the new chunk by taking `min` from the first merged chunk and `max`
+            // from the last.
+            write_ops::UpdateCommandRequest updateOp(ChunkType::ConfigNS);
+            updateOp.setUpdates({[&] {
+                write_ops::UpdateOpEntry entry;
+
+                ChunkType mergedChunk(chunksToMerge->front());
+                entry.setQ(BSON(ChunkType::name(mergedChunk.getName())));
+                mergedChunk.setMax(chunksToMerge->back().getMax());
+
+                // Fill in additional details for sending through transaction.
+                mergedChunk.setVersion(mergeVersion);
+                mergedChunk.setEstimatedSizeBytes(boost::none);
+
+                mergedChunk.setHistory({ChunkHistory(validAfter, mergedChunk.getShard())});
+
+                entry.setU(write_ops::UpdateModification::parseFromClassicUpdate(
+                    mergedChunk.toConfigBSON()));
+                entry.setMulti(false);
+
+                return entry;
+            }()});
+
+            const auto updateRes =
+                writeToConfigDocumentInTxn(opCtx, ChunkType::ConfigNS, updateOp, txnNumber);
+
+            const auto numDocsModified = UpdateOp::parseResponse(updateRes).getN();
+            uassert(ErrorCodes::ConflictingOperationInProgress,
+                    str::stream() << "Unexpected number of modified documents during chunks merge "
+                                     "commit. Modified "
+                                  << numDocsModified << " documents instead of 1",
+                    numDocsModified == 1);
+
+            // Delete the rest of the chunks to be merged.
+            // Remember not to delete the first chunk we're expanding.
+            BSONObjBuilder queryBuilder;
+            queryBuilder << ChunkType::collectionUUID << collectionUUID;
+            queryBuilder << ChunkType::shard(shardId.toString());
+            queryBuilder << ChunkType::min(BSON("$gte" << chunksToMerge->front().getMax()));
+            queryBuilder << ChunkType::min(BSON("$lt" << chunksToMerge->back().getMax()));
+
+            write_ops::DeleteCommandRequest deleteOp(ChunkType::ConfigNS);
+            deleteOp.setDeletes({[&] {
+                write_ops::DeleteOpEntry entry;
+                entry.setQ(queryBuilder.obj());
+                entry.setMulti(true);
+                return entry;
+            }()});
+
+            const auto deleteRes =
+                writeToConfigDocumentInTxn(opCtx, ChunkType::ConfigNS, deleteOp, txnNumber);
+
+            const auto numDocsDeleted = DeleteOp::parseResponse(deleteRes).getN();
+            const int expectedNumDocsDeleted = chunksToMerge->size() - 1;
+            uassert(ErrorCodes::ConflictingOperationInProgress,
+                    str::stream() << "Unexpected number of deleted documents during chunks merge "
+                                     "commit. Deleted "
+                                  << numDocsDeleted << " documents instead of "
+                                  << expectedNumDocsDeleted,
+                    numDocsDeleted == expectedNumDocsDeleted);
+
+            LOGV2_DEBUG(6583805, 1, "Finished all transaction operations in merge chunk command");
+        });
+}
+
 StatusWith<BSONObj> ShardingCatalogManager::commitChunksMerge(
     OperationContext* opCtx,
     const NamespaceString& nss,
@@ -791,11 +806,11 @@ StatusWith<BSONObj> ShardingCatalogManager::commitChunksMerge(
     const boost::optional<Timestamp>& timestamp,
     const UUID& requestCollectionUUID,
     const ChunkRange& chunkRange,
-    const ShardId& shardId,
-    const boost::optional<Timestamp>& validAfter) {
-    if (!validAfter) {
-        return {ErrorCodes::IllegalOperation, "chunk operation requires validAfter timestamp"};
-    }
+    const ShardId& shardId) {
+
+    // Mark opCtx as interruptible to ensure that all reads and writes to the metadata collections
+    // under the exclusive _kChunkOpLock happen on the same term.
+    opCtx->setAlwaysInterruptAtStepDownOrUp();
 
     // Take _kChunkOpLock in exclusive mode to prevent concurrent chunk modifications and generate
     // strictly monotonously increasing collection versions
@@ -871,11 +886,18 @@ StatusWith<BSONObj> ShardingCatalogManager::commitChunksMerge(
 
     // 3. Prepare the data for the merge
     //    and ensure that the retrieved list of chunks covers the whole range.
-    std::vector<ChunkType> chunksToMerge;
+
+    // The `validAfter` field must always be set. If not existing, it means the chunk
+    // always belonged to the same shard, hence it's valid to set `0` as the time at
+    // which the chunk started being valid.
+    Timestamp validAfter{0};
+
+    auto chunksToMerge = std::make_shared<std::vector<ChunkType>>();
+    chunksToMerge->reserve(shardChunksInRangeResponse.docs.size());
     for (const auto& chunkDoc : shardChunksInRangeResponse.docs) {
         auto chunk = uassertStatusOK(
             ChunkType::parseFromConfigBSON(chunkDoc, coll.getEpoch(), coll.getTimestamp()));
-        if (chunksToMerge.empty()) {
+        if (chunksToMerge->empty()) {
             uassert(ErrorCodes::IllegalOperation,
                     str::stream()
                         << "could not merge chunks, shard " << shardId
@@ -888,46 +910,41 @@ StatusWith<BSONObj> ShardingCatalogManager::commitChunksMerge(
                         << "could not merge chunks, shard " << shardId
                         << " does not contain a sequence of chunks that exactly fills the range "
                         << chunkRange.toString(),
-                    chunk.getMin().woCompare(chunksToMerge.back().getMax()) == 0);
+                    chunk.getMin().woCompare(chunksToMerge->back().getMax()) == 0);
         }
-        chunksToMerge.push_back(std::move(chunk));
+
+        // Get the `validAfter` field from the most recent chunk placed on the shard
+        if (!chunk.getHistory().empty()) {
+            const auto& chunkValidAfter = chunk.getHistory().front().getValidAfter();
+            if (validAfter < chunkValidAfter) {
+                validAfter = chunkValidAfter;
+            }
+        }
+
+        chunksToMerge->push_back(std::move(chunk));
     }
     uassert(ErrorCodes::IllegalOperation,
             str::stream() << "could not merge chunks, shard " << shardId
                           << " does not contain a sequence of chunks that exactly fills the range "
                           << chunkRange.toString(),
-            !chunksToMerge.empty() &&
-                chunksToMerge.back().getMax().woCompare(chunkRange.getMax()) == 0);
+            !chunksToMerge->empty() &&
+                chunksToMerge->back().getMax().woCompare(chunkRange.getMax()) == 0);
 
     ChunkVersion initialVersion = collVersion;
     ChunkVersion mergeVersion = initialVersion;
     mergeVersion.incMinor();
 
-    auto updates = buildMergeChunksTransactionUpdates(chunksToMerge, mergeVersion, validAfter);
-    auto preCond = buildMergeChunksTransactionPrecond(chunksToMerge, initialVersion);
-
     // 4. apply the batch of updates to local metadata
-    uassertStatusOK(Grid::get(opCtx)->catalogClient()->applyChunkOpsDeprecated(
-        opCtx,
-        updates,
-        preCond,
-        coll.getUuid(),
-        nss,
-        mergeVersion,
-        WriteConcernOptions(),
-        repl::ReadConcernLevel::kLocalReadConcern));
+    _mergeChunksInTransaction(
+        opCtx, nss, coll.getUuid(), mergeVersion, validAfter, chunkRange, shardId, chunksToMerge);
 
     // 5. log changes
     BSONObjBuilder logDetail;
-    {
-        BSONArrayBuilder b(logDetail.subarrayStart("merged"));
-        for (const auto& chunkToMerge : chunksToMerge) {
-            b.append(chunkToMerge.toConfigBSON());
-        }
-    }
     initialVersion.appendLegacyWithField(&logDetail, "prevShardVersion");
     mergeVersion.appendLegacyWithField(&logDetail, "mergedVersion");
     logDetail.append("owningShard", shardId);
+    chunkRange.append(&logDetail);
+    logDetail.append("numChunks", static_cast<int>(chunksToMerge->size()));
 
     ShardingLogging::get(opCtx)->logChange(
         opCtx, "merge", nss.ns(), logDetail.obj(), WriteConcernOptions());
@@ -952,6 +969,15 @@ StatusWith<BSONObj> ShardingCatalogManager::commitChunkMigration(
         return {ErrorCodes::IllegalOperation, "chunk operation requires validAfter timestamp"};
     }
 
+    uassertStatusOK(
+        ShardKeyPattern::checkShardKeyIsValidForMetadataStorage(migratedChunk.getMin()));
+    uassertStatusOK(
+        ShardKeyPattern::checkShardKeyIsValidForMetadataStorage(migratedChunk.getMax()));
+
+    // Mark opCtx as interruptible to ensure that all reads and writes to the metadata collections
+    // under the exclusive _kChunkOpLock happen on the same term.
+    opCtx->setAlwaysInterruptAtStepDownOrUp();
+
     // Must hold the shard lock until the entire commit finishes to serialize with removeShard.
     Lock::SharedLock shardLock(opCtx->lockState(), _kShardMembershipLock);
 
@@ -1041,12 +1067,8 @@ StatusWith<BSONObj> ShardingCatalogManager::commitChunkMigration(
             migratedChunk.isVersionSet() && migratedChunk.getVersion().isSet());
 
     // Check if range still exists and which shard owns it
-    auto swCurrentChunk = findChunkContainingRange(opCtx,
-                                                   coll.getUuid(),
-                                                   coll.getEpoch(),
-                                                   coll.getTimestamp(),
-                                                   migratedChunk.getMin(),
-                                                   migratedChunk.getMax());
+    auto swCurrentChunk = findChunkContainingRange(
+        opCtx, coll.getUuid(), coll.getEpoch(), coll.getTimestamp(), migratedChunk.getRange());
 
     if (!swCurrentChunk.isOK()) {
         return swCurrentChunk.getStatus();
@@ -1263,6 +1285,10 @@ void ShardingCatalogManager::upgradeChunksHistory(OperationContext* opCtx,
     auto const catalogClient = Grid::get(opCtx)->catalogClient();
     const auto shardRegistry = Grid::get(opCtx)->shardRegistry();
 
+    // Mark opCtx as interruptible to ensure that all reads and writes to the metadata collections
+    // under the exclusive _kChunkOpLock happen on the same term.
+    opCtx->setAlwaysInterruptAtStepDownOrUp();
+
     // Take _kChunkOpLock in exclusive mode to prevent concurrent chunk splits, merges, and
     // migrations.
     Lock::ExclusiveLock lk(opCtx->lockState(), _kChunkOpLock);
@@ -1395,6 +1421,10 @@ void ShardingCatalogManager::clearJumboFlag(OperationContext* opCtx,
                                             const NamespaceString& nss,
                                             const OID& collectionEpoch,
                                             const ChunkRange& chunk) {
+    // Mark opCtx as interruptible to ensure that all reads and writes to the metadata collections
+    // under the exclusive _kChunkOpLock happen on the same term.
+    opCtx->setAlwaysInterruptAtStepDownOrUp();
+
     // Take _kChunkOpLock in exclusive mode to prevent concurrent chunk modifications and generate
     // strictly monotonously increasing collection versions
     Lock::ExclusiveLock lk(opCtx, opCtx->lockState(), _kChunkOpLock);
@@ -1513,6 +1543,10 @@ void ShardingCatalogManager::ensureChunkVersionIsGreaterThan(OperationContext* o
                                                              const BSONObj& minKey,
                                                              const BSONObj& maxKey,
                                                              const ChunkVersion& version) {
+    // Mark opCtx as interruptible to ensure that all reads and writes to the metadata collections
+    // under the exclusive _kChunkOpLock happen on the same term.
+    opCtx->setAlwaysInterruptAtStepDownOrUp();
+
     // Take _kChunkOpLock in exclusive mode to prevent concurrent chunk modifications and generate
     // strictly monotonously increasing collection versions
     Lock::ExclusiveLock lk(opCtx, opCtx->lockState(), _kChunkOpLock);
@@ -1710,36 +1744,58 @@ void ShardingCatalogManager::bumpMultipleCollectionVersionsAndChangeMetadataInTx
     unique_function<void(OperationContext*, TxnNumber)> changeMetadataFunc,
     const WriteConcernOptions& writeConcern) {
 
+    // Mark opCtx as interruptible to ensure that all reads and writes to the metadata collections
+    // under the exclusive _kChunkOpLock happen on the same term.
+    opCtx->setAlwaysInterruptAtStepDownOrUp();
+
     // Take _kChunkOpLock in exclusive mode to prevent concurrent chunk splits, merges, and
     // migrations
     Lock::ExclusiveLock lk(opCtx, opCtx->lockState(), _kChunkOpLock);
 
-    withTransaction(opCtx,
-                    NamespaceString::kConfigReshardingOperationsNamespace,
-                    [&](OperationContext* opCtx, TxnNumber txnNumber) {
-                        for (const auto& nss : collNames) {
-                            bumpCollectionMinorVersion(opCtx, nss, txnNumber);
-                        }
-                        changeMetadataFunc(opCtx, txnNumber);
-                    },
-                    writeConcern);
+    withTransaction(
+        opCtx,
+        NamespaceString::kConfigReshardingOperationsNamespace,
+        [&collNames, &changeMetadataFunc](OperationContext* opCtx, TxnNumber txnNumber) {
+            for (const auto& nss : collNames) {
+                bumpCollectionMinorVersion(opCtx, nss, txnNumber);
+            }
+            changeMetadataFunc(opCtx, txnNumber);
+        },
+        writeConcern);
 }
 
 void ShardingCatalogManager::splitOrMarkJumbo(OperationContext* opCtx,
                                               const NamespaceString& nss,
-                                              const BSONObj& minKey) {
+                                              const BSONObj& minKey,
+                                              boost::optional<int64_t> optMaxChunkSizeBytes) {
     const auto cm = uassertStatusOK(
         Grid::get(opCtx)->catalogCache()->getShardedCollectionRoutingInfoWithRefresh(opCtx, nss));
     auto chunk = cm.findIntersectingChunkWithSimpleCollation(minKey);
 
     try {
-        const auto splitPoints = uassertStatusOK(shardutil::selectChunkSplitPoints(
-            opCtx,
-            chunk.getShardId(),
-            nss,
-            cm.getShardKeyPattern(),
-            ChunkRange(chunk.getMin(), chunk.getMax()),
-            Grid::get(opCtx)->getBalancerConfiguration()->getMaxChunkSizeBytes()));
+        const auto maxChunkSizeBytes = [&]() -> int64_t {
+            if (optMaxChunkSizeBytes.has_value()) {
+                return *optMaxChunkSizeBytes;
+            }
+
+            auto coll = Grid::get(opCtx)->catalogClient()->getCollection(
+                opCtx, nss, repl::ReadConcernLevel::kMajorityReadConcern);
+            return coll.getMaxChunkSizeBytes().value_or(
+                Grid::get(opCtx)->getBalancerConfiguration()->getMaxChunkSizeBytes());
+        }();
+
+        // Limit the search to one split point: this code path is reached when a migration fails due
+        // to ErrorCodes::ChunkTooBig. In case there is a too frequent shard key, only select the
+        // next key in order to split the range in jumbo chunk + remaining range.
+        const int limit = 1;
+        auto splitPoints = uassertStatusOK(
+            shardutil::selectChunkSplitPoints(opCtx,
+                                              chunk.getShardId(),
+                                              nss,
+                                              cm.getShardKeyPattern(),
+                                              ChunkRange(chunk.getMin(), chunk.getMax()),
+                                              maxChunkSizeBytes,
+                                              limit));
 
         if (splitPoints.empty()) {
             LOGV2(21873,
@@ -1793,6 +1849,9 @@ void ShardingCatalogManager::splitOrMarkJumbo(OperationContext* opCtx,
             return;
         }
 
+        // Resize the vector because in multiversion scenarios the `autoSplitVector` command may end
+        // up ignoring the `limit` parameter and returning the whole list of split points.
+        splitPoints.resize(limit);
         uassertStatusOK(
             shardutil::splitChunkAtMultiplePoints(opCtx,
                                                   chunk.getShardId(),
@@ -1814,6 +1873,10 @@ void ShardingCatalogManager::setAllowMigrationsAndBumpOneChunk(
     bool allowMigrations) {
     std::set<ShardId> shardsIds;
     {
+        // Mark opCtx as interruptible to ensure that all reads and writes to the metadata
+        // collections under the exclusive _kChunkOpLock happen on the same term.
+        opCtx->setAlwaysInterruptAtStepDownOrUp();
+
         // Take _kChunkOpLock in exclusive mode to prevent concurrent chunk splits, merges, and
         // migrations
         Lock::ExclusiveLock lk(opCtx, opCtx->lockState(), _kChunkOpLock);
@@ -1830,7 +1893,10 @@ void ShardingCatalogManager::setAllowMigrationsAndBumpOneChunk(
 
         cm.getAllShardIds(&shardsIds);
         withTransaction(
-            opCtx, CollectionType::ConfigNS, [&](OperationContext* opCtx, TxnNumber txnNumber) {
+            opCtx,
+            CollectionType::ConfigNS,
+            [this, allowMigrations, &nss, &collectionUUID](OperationContext* opCtx,
+                                                           TxnNumber txnNumber) {
                 // Update the 'allowMigrations' field. An unset 'allowMigrations' field implies
                 // 'true'. To ease backwards compatibility we omit 'allowMigrations' instead of
                 // setting it explicitly to 'true'.
diff --git a/src/mongo/db/s/config/sharding_catalog_manager_collection_operations.cpp b/src/mongo/db/s/config/sharding_catalog_manager_collection_operations.cpp
index 6b2538bfeca..1cb326b2bb5 100644
--- a/src/mongo/db/s/config/sharding_catalog_manager_collection_operations.cpp
+++ b/src/mongo/db/s/config/sharding_catalog_manager_collection_operations.cpp
@@ -288,6 +288,10 @@ std::pair<std::vector<BSONObj>, std::vector<BSONObj>> makeChunkAndTagUpdatesForR
 void ShardingCatalogManager::refineCollectionShardKey(OperationContext* opCtx,
                                                       const NamespaceString& nss,
                                                       const ShardKeyPattern& newShardKeyPattern) {
+    // Mark opCtx as interruptible to ensure that all reads and writes to the metadata collections
+    // under the exclusive _kChunkOpLock happen on the same term.
+    opCtx->setAlwaysInterruptAtStepDownOrUp();
+
     // Take _kChunkOpLock in exclusive mode to prevent concurrent chunk modifications and generate
     // strictly monotonously increasing collection versions
     Lock::ExclusiveLock chunkLk(opCtx, opCtx->lockState(), _kChunkOpLock);
@@ -323,79 +327,80 @@ void ShardingCatalogManager::refineCollectionShardKey(OperationContext* opCtx,
     Timestamp newTimestamp = now.clusterTime().asTimestamp();
     collType.setTimestamp(newTimestamp);
 
-    auto updateCollectionAndChunksFn = [&](OperationContext* opCtx, TxnNumber txnNumber) {
-        // Update the config.collections entry for the given namespace.
-        updateShardingCatalogEntryForCollectionInTxn(
-            opCtx, nss, collType, false /* upsert */, txnNumber);
-
-        LOGV2(21933,
-              "refineCollectionShardKey updated collection entry for {namespace}: took "
-              "{durationMillis} ms. Total time taken: {totalTimeMillis} ms.",
-              "refineCollectionShardKey updated collection entry",
-              "namespace"_attr = nss.ns(),
-              "durationMillis"_attr = timers->executionTimer.millis(),
-              "totalTimeMillis"_attr = timers->totalTimer.millis());
-        timers->executionTimer.reset();
-
-        if (MONGO_unlikely(hangRefineCollectionShardKeyBeforeUpdatingChunks.shouldFail())) {
-            LOGV2(21934, "Hit hangRefineCollectionShardKeyBeforeUpdatingChunks failpoint");
-            hangRefineCollectionShardKeyBeforeUpdatingChunks.pauseWhileSet(opCtx);
-        }
+    auto updateCollectionAndChunksFn =
+        [this, &nss, &collType, &timers, &newFields](OperationContext* opCtx, TxnNumber txnNumber) {
+            // Update the config.collections entry for the given namespace.
+            updateShardingCatalogEntryForCollectionInTxn(
+                opCtx, nss, collType, false /* upsert */, txnNumber);
+
+            LOGV2(21933,
+                  "refineCollectionShardKey updated collection entry for {namespace}: took "
+                  "{durationMillis} ms. Total time taken: {totalTimeMillis} ms.",
+                  "refineCollectionShardKey updated collection entry",
+                  "namespace"_attr = nss.ns(),
+                  "durationMillis"_attr = timers->executionTimer.millis(),
+                  "totalTimeMillis"_attr = timers->totalTimer.millis());
+            timers->executionTimer.reset();
+
+            if (MONGO_unlikely(hangRefineCollectionShardKeyBeforeUpdatingChunks.shouldFail())) {
+                LOGV2(21934, "Hit hangRefineCollectionShardKeyBeforeUpdatingChunks failpoint");
+                hangRefineCollectionShardKeyBeforeUpdatingChunks.pauseWhileSet(opCtx);
+            }
 
-        auto [chunkUpdates, tagUpdates] = makeChunkAndTagUpdatesForRefine(newFields);
+            auto [chunkUpdates, tagUpdates] = makeChunkAndTagUpdatesForRefine(newFields);
 
-        // Update all config.chunks entries for the given namespace by setting (i) their bounds for
-        // each new field in the refined key to MinKey (except for the global max chunk where the
-        // max bounds are set to MaxKey), and unsetting (ii) their jumbo field.
-        const auto chunksQuery = BSON(ChunkType::collectionUUID << collType.getUuid());
-        writeToConfigDocumentInTxn(
-            opCtx,
-            ChunkType::ConfigNS,
-            BatchedCommandRequest::buildPipelineUpdateOp(ChunkType::ConfigNS,
-                                                         chunksQuery,
-                                                         chunkUpdates,
-                                                         false,  // upsert
-                                                         true    // useMultiUpdate
-                                                         ),
-            txnNumber);
+            // Update all config.chunks entries for the given namespace by setting (i) their bounds
+            // for each new field in the refined key to MinKey (except for the global max chunk
+            // where the max bounds are set to MaxKey), and unsetting (ii) their jumbo field.
+            const auto chunksQuery = BSON(ChunkType::collectionUUID << collType.getUuid());
+            writeToConfigDocumentInTxn(
+                opCtx,
+                ChunkType::ConfigNS,
+                BatchedCommandRequest::buildPipelineUpdateOp(ChunkType::ConfigNS,
+                                                             chunksQuery,
+                                                             chunkUpdates,
+                                                             false,  // upsert
+                                                             true    // useMultiUpdate
+                                                             ),
+                txnNumber);
 
-        LOGV2(21935,
-              "refineCollectionShardKey: updated chunk entries for {namespace}: took "
-              "{durationMillis} ms. Total time taken: {totalTimeMillis} ms.",
-              "refineCollectionShardKey: updated chunk entries",
-              "namespace"_attr = nss.ns(),
-              "durationMillis"_attr = timers->executionTimer.millis(),
-              "totalTimeMillis"_attr = timers->totalTimer.millis());
-        timers->executionTimer.reset();
-
-        // Update all config.tags entries for the given namespace by setting their bounds for
-        // each new field in the refined key to MinKey (except for the global max tag where the
-        // max bounds are set to MaxKey).
-        writeToConfigDocumentInTxn(
-            opCtx,
-            TagsType::ConfigNS,
-            BatchedCommandRequest::buildPipelineUpdateOp(TagsType::ConfigNS,
-                                                         BSON("ns" << nss.ns()),
-                                                         tagUpdates,
-                                                         false,  // upsert
-                                                         true    // useMultiUpdate
-                                                         ),
-            txnNumber);
+            LOGV2(21935,
+                  "refineCollectionShardKey: updated chunk entries for {namespace}: took "
+                  "{durationMillis} ms. Total time taken: {totalTimeMillis} ms.",
+                  "refineCollectionShardKey: updated chunk entries",
+                  "namespace"_attr = nss.ns(),
+                  "durationMillis"_attr = timers->executionTimer.millis(),
+                  "totalTimeMillis"_attr = timers->totalTimer.millis());
+            timers->executionTimer.reset();
+
+            // Update all config.tags entries for the given namespace by setting their bounds for
+            // each new field in the refined key to MinKey (except for the global max tag where the
+            // max bounds are set to MaxKey).
+            writeToConfigDocumentInTxn(
+                opCtx,
+                TagsType::ConfigNS,
+                BatchedCommandRequest::buildPipelineUpdateOp(TagsType::ConfigNS,
+                                                             BSON("ns" << nss.ns()),
+                                                             tagUpdates,
+                                                             false,  // upsert
+                                                             true    // useMultiUpdate
+                                                             ),
+                txnNumber);
 
 
-        LOGV2(21936,
-              "refineCollectionShardKey: updated zone entries for {namespace}: took "
-              "{durationMillis} ms. Total time taken: {totalTimeMillis} ms.",
-              "refineCollectionShardKey: updated zone entries",
-              "namespace"_attr = nss.ns(),
-              "durationMillis"_attr = timers->executionTimer.millis(),
-              "totalTimeMillis"_attr = timers->totalTimer.millis());
+            LOGV2(21936,
+                  "refineCollectionShardKey: updated zone entries for {namespace}: took "
+                  "{durationMillis} ms. Total time taken: {totalTimeMillis} ms.",
+                  "refineCollectionShardKey: updated zone entries",
+                  "namespace"_attr = nss.ns(),
+                  "durationMillis"_attr = timers->executionTimer.millis(),
+                  "totalTimeMillis"_attr = timers->totalTimer.millis());
 
-        if (MONGO_unlikely(hangRefineCollectionShardKeyBeforeCommit.shouldFail())) {
-            LOGV2(21937, "Hit hangRefineCollectionShardKeyBeforeCommit failpoint");
-            hangRefineCollectionShardKeyBeforeCommit.pauseWhileSet(opCtx);
-        }
-    };
+            if (MONGO_unlikely(hangRefineCollectionShardKeyBeforeCommit.shouldFail())) {
+                LOGV2(21937, "Hit hangRefineCollectionShardKeyBeforeCommit failpoint");
+                hangRefineCollectionShardKeyBeforeCommit.pauseWhileSet(opCtx);
+            }
+        };
 
     auto updateCollectionAndChunksWithAPIFn =
         [collType, newFields, nss, timers](const txn_api::TransactionClient& txnClient,
@@ -558,14 +563,6 @@ void ShardingCatalogManager::configureCollectionBalancing(
     boost::optional<bool> defragmentCollection,
     boost::optional<bool> enableAutoSplitter) {
 
-    // Hold the FCV region to serialize with the setFeatureCompatibilityVersion command
-    FixedFCVRegion fcvRegion(opCtx);
-    uassert(ErrorCodes::IllegalOperation,
-            "_configsvrConfigureCollectionBalancing can only be run when the cluster is in feature "
-            "compatibility versions greater or equal than 5.3.",
-            serverGlobalParams.featureCompatibility.isGreaterThanOrEqualTo(
-                multiversion::FeatureCompatibilityVersion::kVersion_5_3));
-
     uassert(ErrorCodes::InvalidOptions,
             "invalid configure collection balancing update",
             chunkSizeMB || defragmentCollection || enableAutoSplitter);
@@ -575,14 +572,14 @@ void ShardingCatalogManager::configureCollectionBalancing(
     {
         BSONObjBuilder setBuilder(updateCmd.subobjStart("$set"));
         if (chunkSizeMB && *chunkSizeMB != 0) {
-            // verify we got a positive integer in range [1MB, 1GB]
+            auto chunkSizeBytes = static_cast<int64_t>(*chunkSizeMB) * 1024 * 1024;
+            bool withinRange = nss == NamespaceString::kLogicalSessionsNamespace
+                ? (chunkSizeBytes > 0 && chunkSizeBytes <= 1024 * 1024 * 1024)
+                : ChunkSizeSettingsType::checkMaxChunkSizeValid(chunkSizeBytes);
             uassert(ErrorCodes::InvalidOptions,
                     str::stream() << "Chunk size '" << *chunkSizeMB << "' out of range [1MB, 1GB]",
-                    *chunkSizeMB > 0 &&
-                        *chunkSizeMB < std::numeric_limits<int32_t>::max() / (1024 * 1024) &&
-                        ChunkSizeSettingsType::checkMaxChunkSizeValid(*chunkSizeMB * 1024 * 1024));
-            setBuilder.append(CollectionType::kMaxChunkSizeBytesFieldName,
-                              *chunkSizeMB * 1024 * 1024);
+                    withinRange);
+            setBuilder.append(CollectionType::kMaxChunkSizeBytesFieldName, chunkSizeBytes);
             updatedFields++;
         }
         if (defragmentCollection) {
@@ -617,26 +614,27 @@ void ShardingCatalogManager::configureCollectionBalancing(
         // migrations
         Lock::ExclusiveLock lk(opCtx, opCtx->lockState(), _kChunkOpLock);
 
-        withTransaction(
-            opCtx, CollectionType::ConfigNS, [&](OperationContext* opCtx, TxnNumber txnNumber) {
-                const auto query = BSON(CollectionType::kNssFieldName << nss.ns());
-                const auto res = writeToConfigDocumentInTxn(
-                    opCtx,
-                    CollectionType::ConfigNS,
-                    BatchedCommandRequest::buildUpdateOp(CollectionType::ConfigNS,
-                                                         query,
-                                                         update /* update */,
-                                                         false /* upsert */,
-                                                         false /* multi */),
-                    txnNumber);
-                const auto numDocsModified = UpdateOp::parseResponse(res).getN();
-                uassert(ErrorCodes::ConflictingOperationInProgress,
-                        str::stream() << "Expected to match one doc for query " << query
-                                      << " but matched " << numDocsModified,
-                        numDocsModified == 1);
-
-                bumpCollectionMinorVersionInTxn(opCtx, nss, txnNumber);
-            });
+        withTransaction(opCtx,
+                        CollectionType::ConfigNS,
+                        [this, &nss, &update](OperationContext* opCtx, TxnNumber txnNumber) {
+                            const auto query = BSON(CollectionType::kNssFieldName << nss.ns());
+                            const auto res = writeToConfigDocumentInTxn(
+                                opCtx,
+                                CollectionType::ConfigNS,
+                                BatchedCommandRequest::buildUpdateOp(CollectionType::ConfigNS,
+                                                                     query,
+                                                                     update /* update */,
+                                                                     false /* upsert */,
+                                                                     false /* multi */),
+                                txnNumber);
+                            const auto numDocsModified = UpdateOp::parseResponse(res).getN();
+                            uassert(ErrorCodes::NamespaceNotSharded,
+                                    str::stream() << "Expected to match one doc for query " << query
+                                                  << " but matched " << numDocsModified,
+                                    numDocsModified == 1);
+
+                            bumpCollectionMinorVersionInTxn(opCtx, nss, txnNumber);
+                        });
         // Now any migrations that change the list of shards will see the results of the transaction
         // during refresh, so it is safe to release the chunk lock.
     }
@@ -703,7 +701,9 @@ void ShardingCatalogManager::updateTimeSeriesGranularity(OperationContext* opCtx
     cm.getAllShardIds(&shardIds);
 
     withTransaction(
-        opCtx, CollectionType::ConfigNS, [&](OperationContext* opCtx, TxnNumber txnNumber) {
+        opCtx,
+        CollectionType::ConfigNS,
+        [this, &nss, granularity, &shardIds](OperationContext* opCtx, TxnNumber txnNumber) {
             // Update granularity value in config.collections.
             auto granularityFieldName = CollectionType::kTimeseriesFieldsFieldName + "." +
                 TypeCollectionTimeseriesFields::kGranularityFieldName;
diff --git a/src/mongo/db/s/config/sharding_catalog_manager_config_initialization_test.cpp b/src/mongo/db/s/config/sharding_catalog_manager_config_initialization_test.cpp
index 6c95002292b..39c741092f9 100644
--- a/src/mongo/db/s/config/sharding_catalog_manager_config_initialization_test.cpp
+++ b/src/mongo/db/s/config/sharding_catalog_manager_config_initialization_test.cpp
@@ -34,7 +34,7 @@
 
 #include "mongo/bson/json.h"
 #include "mongo/db/catalog_raii.h"
-#include "mongo/db/concurrency/write_conflict_exception.h"
+#include "mongo/db/concurrency/exception_util.h"
 #include "mongo/db/curop.h"
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/operation_context.h"
diff --git a/src/mongo/db/s/config/sharding_catalog_manager_database_operations.cpp b/src/mongo/db/s/config/sharding_catalog_manager_database_operations.cpp
index 934eca19e8e..6a8e082b77e 100644
--- a/src/mongo/db/s/config/sharding_catalog_manager_database_operations.cpp
+++ b/src/mongo/db/s/config/sharding_catalog_manager_database_operations.cpp
@@ -37,6 +37,7 @@
 #include "mongo/db/dbdirectclient.h"
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/ops/write_ops.h"
+#include "mongo/db/persistent_task_store.h"
 #include "mongo/db/repl/repl_client_info.h"
 #include "mongo/db/s/dist_lock_manager.h"
 #include "mongo/db/server_options.h"
@@ -93,9 +94,14 @@ DatabaseType ShardingCatalogManager::createDatabase(OperationContext* opCtx,
             dbName.toString(), ShardId::kConfigServerId, DatabaseVersion::makeFixed());
     }
 
+    // It is not allowed to create the 'admin' or 'local' databases, including any alternative
+    // casing. It is allowed to create the 'config' database (handled by the early return above),
+    // but only with that exact casing.
     uassert(ErrorCodes::InvalidOptions,
-            str::stream() << "Cannot manually create database'" << dbName << "'",
-            dbName != NamespaceString::kAdminDb && dbName != NamespaceString::kLocalDb);
+            str::stream() << "Cannot manually create database '" << dbName << "'",
+            !dbName.equalCaseInsensitive(NamespaceString::kAdminDb) &&
+                !dbName.equalCaseInsensitive(NamespaceString::kLocalDb) &&
+                !dbName.equalCaseInsensitive(NamespaceString::kConfigDb));
 
     uassert(ErrorCodes::InvalidNamespace,
             str::stream() << "Invalid db name specified: " << dbName,
@@ -162,7 +168,7 @@ DatabaseType ShardingCatalogManager::createDatabase(OperationContext* opCtx,
         // Do another loop, with the db lock held in order to avoid taking the expensive path on
         // concurrent create database operations
         dbLock.emplace(DistLockManager::get(opCtx)->lockDirectLocally(
-            opCtx, dbName, DistLockManager::kDefaultLockTimeout));
+            opCtx, str::toLower(dbName), DistLockManager::kDefaultLockTimeout));
     }
 
     // Expensive createDatabase code path
@@ -268,4 +274,63 @@ DatabaseType ShardingCatalogManager::createDatabase(OperationContext* opCtx,
     return database;
 }
 
+void ShardingCatalogManager::commitMovePrimary(OperationContext* opCtx,
+                                               const StringData& dbName,
+                                               const DatabaseVersion& expectedDbVersion,
+                                               const ShardId& toShardId) {
+    // Hold the shard lock until the entire commit finishes to serialize with removeShard.
+    Lock::SharedLock shardLock(opCtx->lockState(), _kShardMembershipLock);
+
+    const auto toShardDoc = [&] {
+        DBDirectClient dbClient(opCtx);
+        return dbClient.findOne(NamespaceString::kConfigsvrShardsNamespace,
+                                BSON(ShardType::name << toShardId));
+    }();
+    uassert(ErrorCodes::ShardNotFound,
+            "Requested primary shard {} does not exist"_format(toShardId.toString()),
+            !toShardDoc.isEmpty());
+
+    const auto toShardEntry = uassertStatusOK(ShardType::fromBSON(toShardDoc));
+    uassert(ErrorCodes::ShardNotFound,
+            "Requested primary shard {} is draining"_format(toShardId.toString()),
+            !toShardEntry.getDraining());
+
+    const auto updateOp = [&] {
+        const auto query = [&] {
+            BSONObjBuilder bsonBuilder;
+            bsonBuilder.append(DatabaseType::kNameFieldName, dbName);
+            // Include the version in the update filter to be resilient to potential network retries
+            // and delayed messages.
+            for (const auto [fieldName, fieldValue] : expectedDbVersion.toBSON()) {
+                const auto dottedFieldName = DatabaseType::kVersionFieldName + "." + fieldName;
+                bsonBuilder.appendAs(fieldValue, dottedFieldName);
+            }
+            return bsonBuilder.obj();
+        }();
+
+        const auto update = [&] {
+            const auto newDbVersion = expectedDbVersion.makeUpdated();
+
+            BSONObjBuilder bsonBuilder;
+            bsonBuilder.append(DatabaseType::kPrimaryFieldName, toShardId);
+            bsonBuilder.append(DatabaseType::kVersionFieldName, newDbVersion.toBSON());
+            return BSON("$set" << bsonBuilder.obj());
+        }();
+
+        write_ops::UpdateCommandRequest updateOp(NamespaceString::kConfigDatabasesNamespace);
+        updateOp.setUpdates({[&] {
+            write_ops::UpdateOpEntry entry;
+            entry.setQ(query);
+            entry.setU(write_ops::UpdateModification::parseFromClassicUpdate(update));
+            return entry;
+        }()});
+
+        return updateOp;
+    }();
+
+    DBDirectClient dbClient(opCtx);
+    const auto commandResponse = dbClient.runCommand(updateOp.serialize({}));
+    uassertStatusOK(getStatusFromWriteCommandReply(commandResponse->getCommandReply()));
+}
+
 }  // namespace mongo
diff --git a/src/mongo/db/s/config/sharding_catalog_manager_database_operations_test.cpp b/src/mongo/db/s/config/sharding_catalog_manager_database_operations_test.cpp
new file mode 100644
index 00000000000..029a229d73f
--- /dev/null
+++ b/src/mongo/db/s/config/sharding_catalog_manager_database_operations_test.cpp
@@ -0,0 +1,100 @@
+/**
+ *    Copyright (C) 2023-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#include "mongo/db/s/config/config_server_test_fixture.h"
+#include "mongo/db/s/config/sharding_catalog_manager.h"
+
+namespace mongo {
+namespace {
+
+class ShardingCatalogManagerDatabaseOperationsTest : public ConfigServerTestFixture {
+public:
+    void setUp() override {
+        ConfigServerTestFixture::setUp();
+        _opCtx = operationContext();
+    }
+
+protected:
+    OperationContext* _opCtx;
+};
+
+TEST_F(ShardingCatalogManagerDatabaseOperationsTest, CreateDatabaseAdminFails) {
+    ASSERT_THROWS_CODE(
+        ShardingCatalogManager::get(_opCtx)->createDatabase(_opCtx, "admin"_sd, boost::none),
+        DBException,
+        ErrorCodes::InvalidOptions);
+
+    // Alternative capitalizations are also invalid
+    ASSERT_THROWS_CODE(
+        ShardingCatalogManager::get(_opCtx)->createDatabase(_opCtx, "Admin"_sd, boost::none),
+        DBException,
+        ErrorCodes::InvalidOptions);
+
+    ASSERT_THROWS_CODE(
+        ShardingCatalogManager::get(_opCtx)->createDatabase(_opCtx, "aDmIn"_sd, boost::none),
+        DBException,
+        ErrorCodes::InvalidOptions);
+}
+
+TEST_F(ShardingCatalogManagerDatabaseOperationsTest, CreateDatabaseLocalFails) {
+    ASSERT_THROWS_CODE(
+        ShardingCatalogManager::get(_opCtx)->createDatabase(_opCtx, "local"_sd, boost::none),
+        DBException,
+        ErrorCodes::InvalidOptions);
+
+    // Alternative capitalizations are also invalid
+    ASSERT_THROWS_CODE(
+        ShardingCatalogManager::get(_opCtx)->createDatabase(_opCtx, "Local"_sd, boost::none),
+        DBException,
+        ErrorCodes::InvalidOptions);
+
+    ASSERT_THROWS_CODE(
+        ShardingCatalogManager::get(_opCtx)->createDatabase(_opCtx, "lOcAl"_sd, boost::none),
+        DBException,
+        ErrorCodes::InvalidOptions);
+}
+
+TEST_F(ShardingCatalogManagerDatabaseOperationsTest, CreateDatabaseConfig) {
+    // It is allowed to create the "config" database.
+    ShardingCatalogManager::get(_opCtx)->createDatabase(_opCtx, "config"_sd, boost::none);
+
+    // But alternative capitalizations are invalid.
+    ASSERT_THROWS_CODE(
+        ShardingCatalogManager::get(_opCtx)->createDatabase(_opCtx, "Config"_sd, boost::none),
+        DBException,
+        ErrorCodes::InvalidOptions);
+
+    ASSERT_THROWS_CODE(
+        ShardingCatalogManager::get(_opCtx)->createDatabase(_opCtx, "cOnFiG"_sd, boost::none),
+        DBException,
+        ErrorCodes::InvalidOptions);
+}
+
+}  // namespace
+}  // namespace mongo
diff --git a/src/mongo/db/s/config/sharding_catalog_manager_merge_chunks_test.cpp b/src/mongo/db/s/config/sharding_catalog_manager_merge_chunks_test.cpp
index d8a006c2286..be08bede01c 100644
--- a/src/mongo/db/s/config/sharding_catalog_manager_merge_chunks_test.cpp
+++ b/src/mongo/db/s/config/sharding_catalog_manager_merge_chunks_test.cpp
@@ -30,9 +30,15 @@
 #include "mongo/platform/basic.h"
 
 #include "mongo/client/read_preference.h"
+#include "mongo/db/dbdirectclient.h"
+#include "mongo/db/logical_session_cache_noop.h"
 #include "mongo/db/namespace_string.h"
+#include "mongo/db/read_write_concern_defaults.h"
+#include "mongo/db/read_write_concern_defaults_cache_lookup_mock.h"
 #include "mongo/db/s/config/config_server_test_fixture.h"
 #include "mongo/db/s/config/sharding_catalog_manager.h"
+#include "mongo/db/s/transaction_coordinator_service.h"
+#include "mongo/db/session_catalog_mongod.h"
 #include "mongo/s/catalog/sharding_catalog_client.h"
 #include "mongo/s/catalog/type_chunk.h"
 
@@ -44,18 +50,37 @@ using unittest::assertGet;
 class MergeChunkTest : public ConfigServerTestFixture {
 protected:
     std::string _shardName = "shard0000";
+
     void setUp() override {
         ConfigServerTestFixture::setUp();
+
         ShardType shard;
         shard.setName(_shardName);
         shard.setHost(_shardName + ":12");
         setupShards({shard});
+
+        DBDirectClient client(operationContext());
+        client.createCollection(NamespaceString::kSessionTransactionsTableNamespace.ns());
+        client.createIndexes(NamespaceString::kSessionTransactionsTableNamespace.ns(),
+                             {MongoDSessionCatalog::getConfigTxnPartialIndexSpec()});
+        client.createCollection(CollectionType::ConfigNS.ns());
+
+        ReadWriteConcernDefaults::create(getServiceContext(), _lookupMock.getFetchDefaultsFn());
+        LogicalSessionCache::set(getServiceContext(), std::make_unique<LogicalSessionCacheNoop>());
+        TransactionCoordinatorService::get(operationContext())
+            ->onShardingInitialization(operationContext(), true);
+    }
+
+    void tearDown() override {
+        TransactionCoordinatorService::get(operationContext())->onStepDown();
+        ConfigServerTestFixture::tearDown();
     }
 
     const ShardId _shardId{_shardName};
     const NamespaceString _nss1{"TestDB.TestColl1"};
     const NamespaceString _nss2{"TestDB.TestColl2"};
     const KeyPattern _keyPattern{BSON("x" << 1)};
+    ReadWriteConcernDefaultsLookupMock _lookupMock;
 };
 
 TEST_F(MergeChunkTest, MergeExistingChunksCorrectlyShouldSucceed) {
@@ -75,6 +100,11 @@ TEST_F(MergeChunkTest, MergeExistingChunksCorrectlyShouldSucceed) {
     auto chunk2(chunk);
     chunk2.setName(OID::gen());
 
+    // set histories
+    chunk.setHistory({ChunkHistory{Timestamp{100, 0}, _shardId}});
+    chunk2.setHistory({ChunkHistory{Timestamp{200, 0}, _shardId}});
+
+    // set boundaries
     auto chunkMin = BSON("a" << 1);
     auto chunkBound = BSON("a" << 5);
     auto chunkMax = BSON("a" << 10);
@@ -87,8 +117,6 @@ TEST_F(MergeChunkTest, MergeExistingChunksCorrectlyShouldSucceed) {
 
     setupCollection(_nss1, _keyPattern, {chunk, chunk2});
 
-    Timestamp validAfter{100, 0};
-
     ChunkRange rangeToBeMerged(chunk.getMin(), chunk2.getMax());
 
     auto versions = assertGet(ShardingCatalogManager::get(operationContext())
@@ -98,8 +126,7 @@ TEST_F(MergeChunkTest, MergeExistingChunksCorrectlyShouldSucceed) {
                                                       collTimestamp,
                                                       collUuid,
                                                       rangeToBeMerged,
-                                                      _shardId,
-                                                      validAfter));
+                                                      _shardId));
 
     auto collVersion = ChunkVersion::fromBSONPositionalOrNewerFormat(versions["collectionVersion"]);
     auto shardVersion = ChunkVersion::fromBSONPositionalOrNewerFormat(versions["shardVersion"]);
@@ -141,7 +168,8 @@ TEST_F(MergeChunkTest, MergeExistingChunksCorrectlyShouldSucceed) {
 
     // Make sure history is there
     ASSERT_EQ(1UL, mergedChunk.getHistory().size());
-    ASSERT_EQ(validAfter, mergedChunk.getHistory().front().getValidAfter());
+    ASSERT_EQ(chunk2.getHistory().front().getValidAfter(),
+              mergedChunk.getHistory().front().getValidAfter());
 }
 
 TEST_F(MergeChunkTest, MergeSeveralChunksCorrectlyShouldSucceed) {
@@ -162,6 +190,11 @@ TEST_F(MergeChunkTest, MergeSeveralChunksCorrectlyShouldSucceed) {
     chunk2.setName(OID::gen());
     chunk3.setName(OID::gen());
 
+    // set histories
+    chunk.setHistory({ChunkHistory{Timestamp{100, 10}, _shardId}});
+    chunk2.setHistory({ChunkHistory{Timestamp{200, 1}, _shardId}});
+    chunk3.setHistory({ChunkHistory{Timestamp{50, 0}, _shardId}});
+
     auto chunkMin = BSON("a" << 1);
     auto chunkBound = BSON("a" << 5);
     auto chunkBound2 = BSON("a" << 7);
@@ -179,8 +212,6 @@ TEST_F(MergeChunkTest, MergeSeveralChunksCorrectlyShouldSucceed) {
     setupCollection(_nss1, _keyPattern, {chunk, chunk2, chunk3});
     ChunkRange rangeToBeMerged(chunk.getMin(), chunk3.getMax());
 
-    Timestamp validAfter{100, 0};
-
     ASSERT_OK(ShardingCatalogManager::get(operationContext())
                   ->commitChunksMerge(operationContext(),
                                       _nss1,
@@ -188,8 +219,7 @@ TEST_F(MergeChunkTest, MergeSeveralChunksCorrectlyShouldSucceed) {
                                       collTimestamp,
                                       collUuid,
                                       rangeToBeMerged,
-                                      _shardId,
-                                      validAfter));
+                                      _shardId));
 
     const auto query BSON(ChunkType::collectionUUID() << collUuid);
     auto findResponse = uassertStatusOK(
@@ -220,7 +250,8 @@ TEST_F(MergeChunkTest, MergeSeveralChunksCorrectlyShouldSucceed) {
 
     // Make sure history is there
     ASSERT_EQ(1UL, mergedChunk.getHistory().size());
-    ASSERT_EQ(validAfter, mergedChunk.getHistory().front().getValidAfter());
+    ASSERT_EQ(chunk2.getHistory().front().getValidAfter(),
+              mergedChunk.getHistory().front().getValidAfter());
 }
 
 TEST_F(MergeChunkTest, NewMergeShouldClaimHighestVersion) {
@@ -241,6 +272,10 @@ TEST_F(MergeChunkTest, NewMergeShouldClaimHighestVersion) {
     auto chunk2(chunk);
     chunk2.setName(OID::gen());
 
+    // set histories
+    chunk.setHistory({ChunkHistory{Timestamp{100, 0}, _shardId}});
+    chunk2.setHistory({ChunkHistory{Timestamp{200, 0}, _shardId}});
+
     auto chunkMin = BSON("a" << 1);
     auto chunkBound = BSON("a" << 5);
     auto chunkMax = BSON("a" << 10);
@@ -263,8 +298,6 @@ TEST_F(MergeChunkTest, NewMergeShouldClaimHighestVersion) {
 
     setupCollection(_nss1, _keyPattern, {chunk, chunk2, otherChunk});
 
-    Timestamp validAfter{100, 0};
-
     ASSERT_OK(ShardingCatalogManager::get(operationContext())
                   ->commitChunksMerge(operationContext(),
                                       _nss1,
@@ -272,8 +305,7 @@ TEST_F(MergeChunkTest, NewMergeShouldClaimHighestVersion) {
                                       collTimestamp,
                                       collUuid,
                                       rangeToBeMerged,
-                                      _shardId,
-                                      validAfter));
+                                      _shardId));
 
     const auto query = BSON(ChunkType::collectionUUID() << collUuid);
     auto findResponse = uassertStatusOK(
@@ -304,7 +336,8 @@ TEST_F(MergeChunkTest, NewMergeShouldClaimHighestVersion) {
 
     // Make sure history is there
     ASSERT_EQ(1UL, mergedChunk.getHistory().size());
-    ASSERT_EQ(validAfter, mergedChunk.getHistory().front().getValidAfter());
+    ASSERT_EQ(chunk2.getHistory().front().getValidAfter(),
+              mergedChunk.getHistory().front().getValidAfter());
 }
 
 TEST_F(MergeChunkTest, MergeLeavesOtherChunksAlone) {
@@ -324,6 +357,10 @@ TEST_F(MergeChunkTest, MergeLeavesOtherChunksAlone) {
     auto chunk2(chunk);
     chunk2.setName(OID::gen());
 
+    // set histories
+    chunk.setHistory({ChunkHistory{Timestamp{100, 5}, shardId}});
+    chunk2.setHistory({ChunkHistory{Timestamp{200, 1}, shardId}});
+
     auto chunkMin = BSON("a" << 1);
     auto chunkBound = BSON("a" << 5);
     auto chunkMax = BSON("a" << 10);
@@ -354,8 +391,7 @@ TEST_F(MergeChunkTest, MergeLeavesOtherChunksAlone) {
                                       collTimestamp,
                                       collUuid,
                                       rangeToBeMerged,
-                                      shardId,
-                                      validAfter));
+                                      shardId));
     const auto query = BSON(ChunkType::collectionUUID() << collUuid);
     auto findResponse = uassertStatusOK(
         getConfigShard()->exhaustiveFindOnConfig(operationContext(),
@@ -398,11 +434,16 @@ TEST_F(MergeChunkTest, NonExistingNamespace) {
     chunk.setCollectionUUID(UUID::gen());
 
     auto origVersion = ChunkVersion(1, 0, collEpoch, collTimestamp);
+    chunk.setShard(_shardId);
     chunk.setVersion(origVersion);
 
     // Construct chunk to be merged
     auto chunk2(chunk);
 
+    // set history
+    chunk.setHistory({ChunkHistory{Timestamp{100, 0}, _shardId}});
+    chunk2.setHistory({ChunkHistory{Timestamp{200, 0}, _shardId}});
+
     auto chunkMin = BSON("a" << 1);
     auto chunkBound = BSON("a" << 5);
     auto chunkMax = BSON("a" << 10);
@@ -417,8 +458,6 @@ TEST_F(MergeChunkTest, NonExistingNamespace) {
 
     setupCollection(_nss1, _keyPattern, {chunk, chunk2});
 
-    Timestamp validAfter{1};
-
     ASSERT_THROWS(ShardingCatalogManager::get(operationContext())
                       ->commitChunksMerge(operationContext(),
                                           NamespaceString("TestDB.NonExistingColl"),
@@ -426,8 +465,7 @@ TEST_F(MergeChunkTest, NonExistingNamespace) {
                                           collTimestamp,
                                           collUuidAtRequest,
                                           rangeToBeMerged,
-                                          _shardId,
-                                          validAfter),
+                                          _shardId),
                   DBException);
 }
 
@@ -446,6 +484,10 @@ TEST_F(MergeChunkTest, NonMatchingUUIDsOfChunkAndRequestErrors) {
     // Construct chunk to be merged
     auto chunk2(chunk);
 
+    // set histories
+    chunk.setHistory({ChunkHistory{Timestamp{100, 0}, _shardId}});
+    chunk2.setHistory({ChunkHistory{Timestamp{200, 0}, _shardId}});
+
     auto chunkMin = BSON("a" << 1);
     auto chunkBound = BSON("a" << 5);
     auto chunkMax = BSON("a" << 10);
@@ -460,8 +502,6 @@ TEST_F(MergeChunkTest, NonMatchingUUIDsOfChunkAndRequestErrors) {
 
     setupCollection(_nss1, _keyPattern, {chunk, chunk2});
 
-    Timestamp validAfter{1};
-
     auto mergeStatus = ShardingCatalogManager::get(operationContext())
                            ->commitChunksMerge(operationContext(),
                                                _nss1,
@@ -469,8 +509,7 @@ TEST_F(MergeChunkTest, NonMatchingUUIDsOfChunkAndRequestErrors) {
                                                collTimestamp,
                                                requestUuid,
                                                rangeToBeMerged,
-                                               _shardId,
-                                               validAfter);
+                                               _shardId);
     ASSERT_EQ(ErrorCodes::InvalidUUID, mergeStatus);
 }
 
@@ -494,12 +533,11 @@ TEST_F(MergeChunkTest, MergeAlreadyHappenedSucceeds) {
     mergedChunk.setName(OID::gen());
     mergedChunk.setCollectionUUID(collUuid);
     mergedChunk.setShard(_shardId);
+    mergedChunk.setHistory({ChunkHistory{Timestamp{100, 0}, _shardId}});
 
 
     setupCollection(_nss1, _keyPattern, {mergedChunk});
 
-    Timestamp validAfter{1};
-
     ASSERT_OK(ShardingCatalogManager::get(operationContext())
                   ->commitChunksMerge(operationContext(),
                                       _nss1,
@@ -507,8 +545,7 @@ TEST_F(MergeChunkTest, MergeAlreadyHappenedSucceeds) {
                                       collTimestamp,
                                       collUuid,
                                       rangeToBeMerged,
-                                      _shardId,
-                                      validAfter));
+                                      _shardId));
 
     // Verify that no change to config.chunks happened.
     const auto query = BSON(ChunkType::collectionUUID() << collUuid);
@@ -555,6 +592,11 @@ TEST_F(MergeChunkTest, MergingChunksWithDollarPrefixShouldSucceed) {
     auto chunkBound2 = BSON("a" << BSON("$mixKey" << 1));
     auto chunkMax = BSON("a" << kMaxBSONKey);
 
+    // set histories
+    chunk1.setHistory({ChunkHistory{Timestamp{100, 9}, _shardId}});
+    chunk2.setHistory({ChunkHistory{Timestamp{200, 5}, _shardId}});
+    chunk3.setHistory({ChunkHistory{Timestamp{156, 1}, _shardId}});
+
     // first chunk boundaries
     chunk1.setMin(chunkMin);
     chunk1.setMax(chunkBound1);
@@ -569,7 +611,6 @@ TEST_F(MergeChunkTest, MergingChunksWithDollarPrefixShouldSucceed) {
 
     // Record chunk boundaries for passing into commitChunksMerge
     ChunkRange rangeToBeMerged(chunk1.getMin(), chunk3.getMax());
-    Timestamp validAfter{100, 0};
 
     ASSERT_OK(ShardingCatalogManager::get(operationContext())
                   ->commitChunksMerge(operationContext(),
@@ -578,8 +619,7 @@ TEST_F(MergeChunkTest, MergingChunksWithDollarPrefixShouldSucceed) {
                                       collTimestamp,
                                       collUuid,
                                       rangeToBeMerged,
-                                      _shardId,
-                                      validAfter));
+                                      _shardId));
 
     const auto query = BSON(ChunkType::collectionUUID() << collUuid);
     auto findResponse = uassertStatusOK(
@@ -610,7 +650,8 @@ TEST_F(MergeChunkTest, MergingChunksWithDollarPrefixShouldSucceed) {
 
     // Make sure history is there
     ASSERT_EQ(1UL, mergedChunk.getHistory().size());
-    ASSERT_EQ(validAfter, mergedChunk.getHistory().front().getValidAfter());
+    ASSERT_EQ(chunk2.getHistory().front().getValidAfter(),
+              mergedChunk.getHistory().front().getValidAfter());
 }
 
 }  // namespace
diff --git a/src/mongo/db/s/config/sharding_catalog_manager_shard_operations.cpp b/src/mongo/db/s/config/sharding_catalog_manager_shard_operations.cpp
index da32d388438..16c2bde5d02 100644
--- a/src/mongo/db/s/config/sharding_catalog_manager_shard_operations.cpp
+++ b/src/mongo/db/s/config/sharding_catalog_manager_shard_operations.cpp
@@ -41,7 +41,6 @@
 #include "mongo/bson/bsonobj_comparator.h"
 #include "mongo/bson/util/bson_extract.h"
 #include "mongo/client/connection_string.h"
-#include "mongo/client/fetcher.h"
 #include "mongo/client/read_preference.h"
 #include "mongo/client/remote_command_targeter.h"
 #include "mongo/client/replica_set_monitor.h"
@@ -54,6 +53,7 @@
 #include "mongo/db/commands/feature_compatibility_version_parser.h"
 #include "mongo/db/commands/set_cluster_parameter_invocation.h"
 #include "mongo/db/commands/set_feature_compatibility_version_gen.h"
+#include "mongo/db/keys_collection_util.h"
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/operation_context.h"
 #include "mongo/db/persistent_task_store.h"
@@ -61,8 +61,10 @@
 #include "mongo/db/repl/hello_gen.h"
 #include "mongo/db/repl/repl_client_info.h"
 #include "mongo/db/repl/repl_set_config.h"
+#include "mongo/db/repl/wait_for_majority_service.h"
 #include "mongo/db/s/add_shard_cmd_gen.h"
 #include "mongo/db/s/add_shard_util.h"
+#include "mongo/db/s/sharding_config_server_parameters_gen.h"
 #include "mongo/db/s/sharding_logging.h"
 #include "mongo/db/s/type_shard_identity.h"
 #include "mongo/db/s/user_writes_critical_section_document_gen.h"
@@ -106,6 +108,8 @@ const WriteConcernOptions kMajorityWriteConcern{WriteConcernOptions::kMajority,
                                                 WriteConcernOptions::SyncMode::UNSET,
                                                 WriteConcernOptions::kNoTimeout};
 
+const Seconds kRemoteCommandTimeout{60};
+
 /**
  * Generates a unique name to be given to a newly added shard.
  */
@@ -153,7 +157,7 @@ StatusWith<std::string> generateNewShardName(OperationContext* opCtx) {
 
 StatusWith<Shard::CommandResponse> ShardingCatalogManager::_runCommandForAddShard(
     OperationContext* opCtx,
-    RemoteCommandTargeter* targeter,
+    std::shared_ptr<RemoteCommandTargeter> targeter,
     StringData dbName,
     const BSONObj& cmdObj) {
     auto swHost = targeter->findHost(opCtx, ReadPreferenceSetting{ReadPreference::PrimaryOnly});
@@ -163,7 +167,7 @@ StatusWith<Shard::CommandResponse> ShardingCatalogManager::_runCommandForAddShar
     auto host = std::move(swHost.getValue());
 
     executor::RemoteCommandRequest request(
-        host, dbName.toString(), cmdObj, rpc::makeEmptyMetadata(), opCtx, Seconds(60));
+        host, dbName.toString(), cmdObj, rpc::makeEmptyMetadata(), opCtx, kRemoteCommandTimeout);
 
     executor::RemoteCommandResponse response =
         Status(ErrorCodes::InternalError, "Internal error running command");
@@ -319,8 +323,8 @@ StatusWith<ShardType> ShardingCatalogManager::_validateHostAsShard(
     std::shared_ptr<RemoteCommandTargeter> targeter,
     const std::string* shardProposedName,
     const ConnectionString& connectionString) {
-    auto swCommandResponse = _runCommandForAddShard(
-        opCtx, targeter.get(), NamespaceString::kAdminDb, BSON("isMaster" << 1));
+    auto swCommandResponse =
+        _runCommandForAddShard(opCtx, targeter, NamespaceString::kAdminDb, BSON("hello" << 1));
     if (swCommandResponse.getStatus() == ErrorCodes::IncompatibleServerVersion) {
         return swCommandResponse.getStatus().withReason(
             str::stream() << "Cannot add " << connectionString.toString()
@@ -331,17 +335,16 @@ StatusWith<ShardType> ShardingCatalogManager::_validateHostAsShard(
     }
 
     // Check for a command response error
-    auto resIsMasterStatus = std::move(swCommandResponse.getValue().commandStatus);
-    if (!resIsMasterStatus.isOK()) {
-        return resIsMasterStatus.withContext(str::stream()
-                                             << "Error running isMaster against "
-                                             << targeter->connectionString().toString());
+    auto resHelloStatus = std::move(swCommandResponse.getValue().commandStatus);
+    if (!resHelloStatus.isOK()) {
+        return resHelloStatus.withContext(str::stream() << "Error running 'hello' against "
+                                                        << targeter->connectionString().toString());
     }
 
-    auto resIsMaster = std::move(swCommandResponse.getValue().response);
+    auto resHello = std::move(swCommandResponse.getValue().response);
 
     // Fail if the node being added is a mongos.
-    const std::string msg = resIsMaster.getStringField("msg").toString();
+    const std::string msg = resHello.getStringField("msg").toString();
     if (msg == "isdbgrid") {
         return {ErrorCodes::IllegalOperation, "cannot add a mongos as a shard"};
     }
@@ -352,23 +355,23 @@ StatusWith<ShardType> ShardingCatalogManager::_validateHostAsShard(
     // because of our internal wire version protocol. So we can safely invariant here that the node
     // is compatible.
     long long maxWireVersion;
-    Status status = bsonExtractIntegerField(resIsMaster, "maxWireVersion", &maxWireVersion);
+    Status status = bsonExtractIntegerField(resHello, "maxWireVersion", &maxWireVersion);
     if (!status.isOK()) {
-        return status.withContext(str::stream() << "isMaster returned invalid 'maxWireVersion' "
+        return status.withContext(str::stream() << "hello returned invalid 'maxWireVersion' "
                                                 << "field when attempting to add "
                                                 << connectionString.toString() << " as a shard");
     }
 
-    // Check whether there is a master. If there isn't, the replica set may not have been
-    // initiated. If the connection is a standalone, it will return true for isMaster.
-    bool isMaster;
-    status = bsonExtractBooleanField(resIsMaster, "ismaster", &isMaster);
+    // Check whether the host is a writable primary. If not, the replica set may not have been
+    // initiated. If the connection is a standalone, it will return true for "isWritablePrimary".
+    bool isWritablePrimary;
+    status = bsonExtractBooleanField(resHello, "isWritablePrimary", &isWritablePrimary);
     if (!status.isOK()) {
-        return status.withContext(str::stream() << "isMaster returned invalid 'ismaster' "
+        return status.withContext(str::stream() << "hello returned invalid 'isWritablePrimary' "
                                                 << "field when attempting to add "
                                                 << connectionString.toString() << " as a shard");
     }
-    if (!isMaster) {
+    if (!isWritablePrimary) {
         return {ErrorCodes::NotWritablePrimary,
                 str::stream()
                     << connectionString.toString()
@@ -377,7 +380,7 @@ StatusWith<ShardType> ShardingCatalogManager::_validateHostAsShard(
     }
 
     const std::string providedSetName = connectionString.getSetName();
-    const std::string foundSetName = resIsMaster["setName"].str();
+    const std::string foundSetName = resHello["setName"].str();
 
     // Make sure the specified replica set name (if any) matches the actual shard's replica set
     if (providedSetName.empty() && !foundSetName.empty()) {
@@ -390,7 +393,7 @@ StatusWith<ShardType> ShardingCatalogManager::_validateHostAsShard(
     if (!providedSetName.empty() && foundSetName.empty()) {
         return {ErrorCodes::OperationFailed,
                 str::stream() << "host did not return a set name; "
-                              << "is the replica set still initializing? " << resIsMaster};
+                              << "is the replica set still initializing? " << resHello};
     }
 
     // Make sure the set name specified in the connection string matches the one where its hosts
@@ -402,14 +405,14 @@ StatusWith<ShardType> ShardingCatalogManager::_validateHostAsShard(
     }
 
     // Is it a config server?
-    if (resIsMaster.hasField("configsvr")) {
+    if (resHello.hasField("configsvr")) {
         return {ErrorCodes::OperationFailed,
                 str::stream() << "Cannot add " << connectionString.toString()
                               << " as a shard since it is a config server"};
     }
 
-    if (resIsMaster.hasField(HelloCommandReply::kIsImplicitDefaultMajorityWCFieldName) &&
-        !resIsMaster.getBoolField(HelloCommandReply::kIsImplicitDefaultMajorityWCFieldName) &&
+    if (resHello.hasField(HelloCommandReply::kIsImplicitDefaultMajorityWCFieldName) &&
+        !resHello.getBoolField(HelloCommandReply::kIsImplicitDefaultMajorityWCFieldName) &&
         !ReadWriteConcernDefaults::get(opCtx).isCWWCSet(opCtx)) {
         return {
             ErrorCodes::OperationFailed,
@@ -422,11 +425,11 @@ StatusWith<ShardType> ShardingCatalogManager::_validateHostAsShard(
                    "using the setDefaultRWConcern command and try again."};
     }
 
-    if (resIsMaster.hasField(HelloCommandReply::kCwwcFieldName)) {
-        auto cwwcOnShard = WriteConcernOptions::parse(
-                               resIsMaster.getObjectField(HelloCommandReply::kCwwcFieldName))
-                               .getValue()
-                               .toBSON();
+    if (resHello.hasField(HelloCommandReply::kCwwcFieldName)) {
+        auto cwwcOnShard =
+            WriteConcernOptions::parse(resHello.getObjectField(HelloCommandReply::kCwwcFieldName))
+                .getValue()
+                .toBSON();
 
         auto cachedCWWC = ReadWriteConcernDefaults::get(opCtx).getCWWC(opCtx);
         if (!cachedCWWC) {
@@ -459,20 +462,20 @@ StatusWith<ShardType> ShardingCatalogManager::_validateHostAsShard(
     if (!providedSetName.empty()) {
         std::set<std::string> hostSet;
 
-        BSONObjIterator iter(resIsMaster["hosts"].Obj());
+        BSONObjIterator iter(resHello["hosts"].Obj());
         while (iter.more()) {
             hostSet.insert(iter.next().String());  // host:port
         }
 
-        if (resIsMaster["passives"].isABSONObj()) {
-            BSONObjIterator piter(resIsMaster["passives"].Obj());
+        if (resHello["passives"].isABSONObj()) {
+            BSONObjIterator piter(resHello["passives"].Obj());
             while (piter.more()) {
                 hostSet.insert(piter.next().String());  // host:port
             }
         }
 
-        if (resIsMaster["arbiters"].isABSONObj()) {
-            BSONObjIterator piter(resIsMaster["arbiters"].Obj());
+        if (resHello["arbiters"].isABSONObj()) {
+            BSONObjIterator piter(resHello["arbiters"].Obj());
             while (piter.more()) {
                 hostSet.insert(piter.next().String());  // host:port
             }
@@ -484,7 +487,7 @@ StatusWith<ShardType> ShardingCatalogManager::_validateHostAsShard(
                 return {ErrorCodes::OperationFailed,
                         str::stream() << "in seed list " << connectionString.toString() << ", host "
                                       << host << " does not belong to replica set " << foundSetName
-                                      << "; found " << resIsMaster.toString()};
+                                      << "; found " << resHello.toString()};
             }
         }
     }
@@ -526,7 +529,7 @@ Status ShardingCatalogManager::_dropSessionsCollection(
     }
 
     auto swCommandResponse = _runCommandForAddShard(
-        opCtx, targeter.get(), NamespaceString::kLogicalSessionsNamespace.db(), builder.done());
+        opCtx, targeter, NamespaceString::kLogicalSessionsNamespace.db(), builder.done());
     if (!swCommandResponse.isOK()) {
         return swCommandResponse.getStatus();
     }
@@ -544,7 +547,7 @@ StatusWith<std::vector<std::string>> ShardingCatalogManager::_getDBNamesListFrom
 
     auto swCommandResponse =
         _runCommandForAddShard(opCtx,
-                               targeter.get(),
+                               targeter,
                                NamespaceString::kAdminDb,
                                BSON("listDatabases" << 1 << "nameOnly" << true));
     if (!swCommandResponse.isOK()) {
@@ -654,6 +657,11 @@ StatusWith<std::string> ShardingCatalogManager::addShard(
             "collection from the shard manually and try again.");
     }
 
+    auto pullKeysStatus = _pullClusterTimeKeys(opCtx, targeter);
+    if (!pullKeysStatus.isOK()) {
+        return pullKeysStatus;
+    }
+
     // If a name for a shard wasn't provided, generate one
     if (shardType.getName().empty()) {
         auto result = generateNewShardName(opCtx);
@@ -670,7 +678,7 @@ StatusWith<std::string> ShardingCatalogManager::addShard(
     // Helper function that runs a command on the to-be shard and returns the status
     auto runCmdOnNewShard = [this, &opCtx, &targeter](const BSONObj& cmd) -> Status {
         auto swCommandResponse =
-            _runCommandForAddShard(opCtx, targeter.get(), NamespaceString::kAdminDb, cmd);
+            _runCommandForAddShard(opCtx, targeter, NamespaceString::kAdminDb, cmd);
         if (!swCommandResponse.isOK()) {
             return swCommandResponse.getStatus();
         }
@@ -692,10 +700,10 @@ StatusWith<std::string> ShardingCatalogManager::addShard(
     }
 
     // Set the user-writes blocking state on the new shard.
-    _setUserWriteBlockingStateOnNewShard(opCtx, targeter.get());
+    _setUserWriteBlockingStateOnNewShard(opCtx, targeter);
 
     // Determine the set of cluster parameters to be used.
-    _standardizeClusterParameters(opCtx, targeter.get());
+    _standardizeClusterParameters(opCtx, targeter);
 
     {
         // Keep the FCV stable across checking the FCV, sending setFCV to the new shard and writing
@@ -721,7 +729,7 @@ StatusWith<std::string> ShardingCatalogManager::addShard(
 
         auto versionResponse =
             _runCommandForAddShard(opCtx,
-                                   targeter.get(),
+                                   targeter,
                                    NamespaceString::kAdminDb,
                                    setFcvCmd.toBSON(BSON(WriteConcernOptions::kWriteConcernField
                                                          << opCtx->getWriteConcern().toBSON())));
@@ -1059,8 +1067,8 @@ StatusWith<long long> ShardingCatalogManager::_runCountCommandOnConfig(Operation
     return result;
 }
 
-void ShardingCatalogManager::_setUserWriteBlockingStateOnNewShard(OperationContext* opCtx,
-                                                                  RemoteCommandTargeter* targeter) {
+void ShardingCatalogManager::_setUserWriteBlockingStateOnNewShard(
+    OperationContext* opCtx, std::shared_ptr<RemoteCommandTargeter> targeter) {
     // Delete all the config.user_writes_critical_sections documents from the new shard.
     {
         write_ops::DeleteCommandRequest deleteOp(
@@ -1126,6 +1134,109 @@ void ShardingCatalogManager::_setUserWriteBlockingStateOnNewShard(OperationConte
     });
 }
 
+std::unique_ptr<Fetcher> ShardingCatalogManager::_createFetcher(
+    OperationContext* opCtx,
+    std::shared_ptr<RemoteCommandTargeter> targeter,
+    const NamespaceString& nss,
+    const repl::ReadConcernLevel& readConcernLevel,
+    FetcherDocsCallbackFn processDocsCallback,
+    FetcherStatusCallbackFn processStatusCallback) {
+    auto host = uassertStatusOK(
+        targeter->findHost(opCtx, ReadPreferenceSetting{ReadPreference::PrimaryOnly}));
+
+    FindCommandRequest findCommand(nss);
+    const auto readConcern =
+        repl::ReadConcernArgs(boost::optional<repl::ReadConcernLevel>(readConcernLevel));
+    findCommand.setReadConcern(readConcern.toBSONInner());
+    const Milliseconds maxTimeMS =
+        std::min(opCtx->getRemainingMaxTimeMillis(), Milliseconds(kRemoteCommandTimeout));
+    findCommand.setMaxTimeMS(durationCount<Milliseconds>(maxTimeMS));
+
+    auto fetcherCallback = [processDocsCallback,
+                            processStatusCallback](const Fetcher::QueryResponseStatus& dataStatus,
+                                                   Fetcher::NextAction* nextAction,
+                                                   BSONObjBuilder* getMoreBob) {
+        // Throw out any accumulated results on error.
+        if (!dataStatus.isOK()) {
+            processStatusCallback(dataStatus.getStatus());
+            return;
+        }
+        const auto& data = dataStatus.getValue();
+
+        try {
+            if (!processDocsCallback(data.documents)) {
+                *nextAction = Fetcher::NextAction::kNoAction;
+            }
+        } catch (DBException& ex) {
+            processStatusCallback(ex.toStatus());
+            return;
+        }
+        processStatusCallback(Status::OK());
+
+        if (!getMoreBob) {
+            return;
+        }
+        getMoreBob->append("getMore", data.cursorId);
+        getMoreBob->append("collection", data.nss.coll());
+    };
+
+    return std::make_unique<Fetcher>(_executorForAddShard.get(),
+                                     host,
+                                     nss.db().toString(),
+                                     findCommand.toBSON({}),
+                                     fetcherCallback,
+                                     BSONObj(), /* metadata tracking, only used for shards */
+                                     maxTimeMS, /* command network timeout */
+                                     maxTimeMS /* getMore network timeout */);
+}
+
+Status ShardingCatalogManager::_pullClusterTimeKeys(
+    OperationContext* opCtx, std::shared_ptr<RemoteCommandTargeter> targeter) {
+    Status fetchStatus =
+        Status(ErrorCodes::InternalError, "Internal error running cursor callback in command");
+    std::vector<ExternalKeysCollectionDocument> keyDocs;
+
+    auto expireAt = opCtx->getServiceContext()->getFastClockSource()->now() +
+        Seconds(gNewShardExistingClusterTimeKeysExpirationSecs.load());
+    auto fetcher = _createFetcher(
+        opCtx,
+        targeter,
+        NamespaceString::kKeysCollectionNamespace,
+        repl::ReadConcernLevel::kLocalReadConcern,
+        [&](const std::vector<BSONObj>& docs) -> bool {
+            for (const BSONObj& doc : docs) {
+                keyDocs.push_back(keys_collection_util::makeExternalClusterTimeKeyDoc(
+                    doc.getOwned(), boost::none /* migrationId */, expireAt));
+            }
+            return true;
+        },
+        [&](const Status& status) { fetchStatus = status; });
+
+    auto scheduleStatus = fetcher->schedule();
+    if (!scheduleStatus.isOK()) {
+        return scheduleStatus;
+    }
+
+    auto joinStatus = fetcher->join(opCtx);
+    if (!joinStatus.isOK()) {
+        return joinStatus;
+    }
+
+    if (keyDocs.empty()) {
+        return fetchStatus;
+    }
+
+    auto opTime = keys_collection_util::storeExternalClusterTimeKeyDocs(opCtx, std::move(keyDocs));
+    auto waitStatus = WaitForMajorityService::get(opCtx->getServiceContext())
+                          .waitUntilMajority(opTime, opCtx->getCancellationToken())
+                          .getNoThrow();
+    if (!waitStatus.isOK()) {
+        return waitStatus;
+    }
+
+    return fetchStatus;
+}
+
 void ShardingCatalogManager::_setClusterParametersLocally(OperationContext* opCtx,
                                                           const std::vector<BSONObj>& parameters) {
     DBDirectClient client(opCtx);
@@ -1145,78 +1256,40 @@ void ShardingCatalogManager::_setClusterParametersLocally(OperationContext* opCt
     }
 }
 
-void ShardingCatalogManager::_pullClusterParametersFromNewShard(OperationContext* opCtx,
-                                                                RemoteCommandTargeter* targeter) {
+void ShardingCatalogManager::_pullClusterParametersFromNewShard(
+    OperationContext* opCtx, std::shared_ptr<RemoteCommandTargeter> targeter) {
     LOGV2(6538600, "Pulling cluster parameters from new shard");
 
     // We can safely query the cluster parameters because the replica set must have been started
     // with --shardsvr in order to add it into the cluster, and in this mode no setClusterParameter
     // can be called on the replica set directly.
-    auto host = uassertStatusOK(
-        targeter->findHost(opCtx, ReadPreferenceSetting{ReadPreference::PrimaryOnly}));
-
-    const Milliseconds maxTimeMS =
-        std::min(opCtx->getRemainingMaxTimeMillis(), Milliseconds(Seconds{30}));
-    BSONObjBuilder findCmdBuilder;
-    {
-        FindCommandRequest findCommand(NamespaceString::kClusterParametersNamespace);
-        auto readConcern = repl::ReadConcernArgs(
-            boost::optional<repl::ReadConcernLevel>(repl::ReadConcernLevel::kMajorityReadConcern));
-        findCommand.setReadConcern(readConcern.toBSONInner());
-        findCommand.setMaxTimeMS(durationCount<Milliseconds>(maxTimeMS));
-        findCommand.serialize(BSONObj(), &findCmdBuilder);
-    }
 
     // If for some reason the callback never gets invoked, we will return this status in response.
-    Status status =
+    Status fetchStatus =
         Status(ErrorCodes::InternalError, "Internal error running cursor callback in command");
-
     std::vector<BSONObj> parameters;
-    auto fetcherCallback =
-        [this, &status, &parameters](const Fetcher::QueryResponseStatus& dataStatus,
-                                     Fetcher::NextAction* nextAction,
-                                     BSONObjBuilder* getMoreBob) {
-            // Throw out any accumulated results on error
-            if (!dataStatus.isOK()) {
-                status = dataStatus.getStatus();
-                return;
-            }
-            const auto& data = dataStatus.getValue();
-
-            for (const BSONObj& doc : data.documents) {
-                parameters.push_back(doc.getOwned());
-            }
-
-            status = Status::OK();
-
-            if (!getMoreBob) {
-                return;
-            }
-            getMoreBob->append("getMore", data.cursorId);
-            getMoreBob->append("collection", data.nss.coll());
-        };
-
-    Fetcher fetcher(_executorForAddShard.get(),
-                    std::move(host),
-                    NamespaceString::kClusterParametersNamespace.db().toString(),
-                    findCmdBuilder.obj(),
-                    fetcherCallback,
-                    BSONObj(), /* metadata tracking, only used for shards */
-                    maxTimeMS, /* command network timeout */
-                    maxTimeMS /* getMore network timeout */);
-
-    uassertStatusOK(fetcher.schedule());
-
-    uassertStatusOK(fetcher.join(opCtx));
-
-    uassertStatusOK(status);
 
+    auto fetcher = _createFetcher(opCtx,
+                                  targeter,
+                                  NamespaceString::kClusterParametersNamespace,
+                                  repl::ReadConcernLevel::kMajorityReadConcern,
+                                  [&parameters](const std::vector<BSONObj>& docs) -> bool {
+                                      for (const BSONObj& doc : docs) {
+                                          parameters.push_back(doc.getOwned());
+                                      }
+                                      return true;
+                                  },
+                                  [&fetchStatus](const Status& status) { fetchStatus = status; });
+
+    uassertStatusOK(fetcher->schedule());
+    uassertStatusOK(fetcher->join(opCtx));
+    uassertStatusOK(fetchStatus);
     _setClusterParametersLocally(opCtx, parameters);
 }
 
 void ShardingCatalogManager::_pushClusterParametersToNewShard(
     OperationContext* opCtx,
-    RemoteCommandTargeter* targeter,
+    std::shared_ptr<RemoteCommandTargeter> targeter,
     const std::vector<BSONObj>& clusterParameters) {
     LOGV2(6360600, "Pushing cluster parameters into new shard");
 
@@ -1252,8 +1325,8 @@ void ShardingCatalogManager::_pushClusterParametersToNewShard(
     }
 }
 
-void ShardingCatalogManager::_standardizeClusterParameters(OperationContext* opCtx,
-                                                           RemoteCommandTargeter* targeter) {
+void ShardingCatalogManager::_standardizeClusterParameters(
+    OperationContext* opCtx, std::shared_ptr<RemoteCommandTargeter> targeter) {
     if (!gFeatureFlagClusterWideConfig.isEnabled(serverGlobalParams.featureCompatibility))
         return;
 
diff --git a/src/mongo/db/s/config/sharding_catalog_manager_split_chunk_test.cpp b/src/mongo/db/s/config/sharding_catalog_manager_split_chunk_test.cpp
index cd7f46ebfda..967dd335d44 100644
--- a/src/mongo/db/s/config/sharding_catalog_manager_split_chunk_test.cpp
+++ b/src/mongo/db/s/config/sharding_catalog_manager_split_chunk_test.cpp
@@ -617,5 +617,54 @@ TEST_F(SplitChunkTest, CantCommitSplitFromChunkSplitterDuringDefragmentation) {
                                            false /* fromChunkSplitter*/));
 }
 
+TEST_F(SplitChunkTest, SplitJumboChunkShouldUnsetJumboFlag) {
+    const auto& nss = _nss2;
+    const auto collTimestamp = Timestamp(42);
+    const auto collEpoch = OID::gen();
+    const auto collUuid = UUID::gen();
+
+    ChunkType chunk;
+    chunk.setName(OID::gen());
+    chunk.setCollectionUUID(collUuid);
+
+    auto origVersion = ChunkVersion(1, 0, collEpoch, collTimestamp);
+    chunk.setVersion(origVersion);
+    chunk.setShard(ShardId(_shardName));
+    chunk.setJumbo(true);
+
+    auto chunkMin = BSON("a" << 1);
+    auto chunkMax = BSON("a" << 10);
+    chunk.setMin(chunkMin);
+    chunk.setMax(chunkMax);
+
+    auto chunkSplitPoint = BSON("a" << 5);
+    std::vector<BSONObj> splitPoints{chunkSplitPoint};
+
+    setupCollection(nss, _keyPattern, {chunk});
+
+    ASSERT_EQ(true, chunk.getJumbo());
+
+    uassertStatusOK(ShardingCatalogManager::get(operationContext())
+                        ->commitChunkSplit(operationContext(),
+                                           nss,
+                                           collEpoch,
+                                           collTimestamp,
+                                           ChunkRange(chunkMin, chunkMax),
+                                           splitPoints,
+                                           "shard0000",
+                                           false /* fromChunkSplitter*/));
+
+    // Both resulting chunks must not be jumbo
+    auto chunkDocLeft =
+        getChunkDoc(operationContext(), collUuid, chunkMin, collEpoch, collTimestamp);
+    ASSERT_OK(chunkDocLeft.getStatus());
+
+    auto chunkDocRight =
+        getChunkDoc(operationContext(), collUuid, chunkSplitPoint, collEpoch, collTimestamp);
+    ASSERT_OK(chunkDocRight.getStatus());
+
+    ASSERT_EQ(false, chunkDocLeft.getValue().getJumbo());
+    ASSERT_EQ(false, chunkDocRight.getValue().getJumbo());
+}
 }  // namespace
 }  // namespace mongo