Update upstream source from tag 'upstream/6.0.0'master

Update to upstream version '6.0.0'
with Debian dir 5604a80ec1c96ca76f25f40d78e6ef855abec322

1 files changed, 148 insertions, 414 deletions

diff --git a/src/mongo/db/s/balancer/balancer.cpp b/src/mongo/db/s/balancer/balancer.cpp
index cef33998f40..4487672302a 100644
--- a/src/mongo/db/s/balancer/balancer.cpp
+++ b/src/mongo/db/s/balancer/balancer.cpp
@@ -52,7 +52,6 @@
 #include "mongo/db/s/config/sharding_catalog_manager.h"
 #include "mongo/db/s/sharding_config_server_parameters_gen.h"
 #include "mongo/db/s/sharding_logging.h"
-#include "mongo/db/server_feature_flags_gen.h"
 #include "mongo/executor/scoped_task_executor.h"
 #include "mongo/logv2/log.h"
 #include "mongo/s/balancer_configuration.h"
@@ -78,9 +77,14 @@ using std::vector;
 namespace {
 
 MONGO_FAIL_POINT_DEFINE(overrideBalanceRoundInterval);
-MONGO_FAIL_POINT_DEFINE(forceBalancerWarningChecks);
 
-const Milliseconds kBalanceRoundDefaultInterval(10 * 1000);
+const Seconds kBalanceRoundDefaultInterval(10);
+
+// Sleep between balancer rounds in the case where the last round found some chunks which needed to
+// be balanced. This value should be set sufficiently low so that imbalanced clusters will quickly
+// reach balanced state, but setting it too low may cause CRUD operations to start failing due to
+// not being able to establish a stable shard version.
+const Seconds kShortBalanceRoundInterval(1);
 
 /**
  * Balancer status response
@@ -97,19 +101,10 @@ class BalanceRoundDetails {
 public:
     BalanceRoundDetails() : _executionTimer() {}
 
-    void setSucceeded(int numCandidateChunks,
-                      int numChunksMoved,
-                      int numImbalancedCachedCollections,
-                      Milliseconds selectionTime,
-                      Milliseconds throttleTime,
-                      Milliseconds migrationTime) {
+    void setSucceeded(int candidateChunks, int chunksMoved) {
         invariant(!_errMsg);
-        _numCandidateChunks = numCandidateChunks;
-        _numChunksMoved = numChunksMoved;
-        _numImbalancedCachedCollections = numImbalancedCachedCollections;
-        _selectionTime = selectionTime;
-        _throttleTime = throttleTime;
-        _migrationTime = migrationTime;
+        _candidateChunks = candidateChunks;
+        _chunksMoved = chunksMoved;
     }
 
     void setFailed(const string& errMsg) {
@@ -124,33 +119,52 @@ public:
         if (_errMsg) {
             builder.append("errmsg", *_errMsg);
         } else {
-            builder.append("candidateChunks", _numCandidateChunks);
-            builder.append("chunksMoved", _numChunksMoved);
-            builder.append("imbalancedCachedCollections", _numImbalancedCachedCollections);
-            BSONObjBuilder timeInfo{builder.subobjStart("times"_sd)};
-            timeInfo.append("selectionTimeMillis"_sd, _selectionTime.count());
-            timeInfo.append("throttleTimeMillis"_sd, _throttleTime.count());
-            timeInfo.append("migrationTimeMillis"_sd, _migrationTime.count());
-            timeInfo.done();
+            builder.append("candidateChunks", _candidateChunks);
+            builder.append("chunksMoved", _chunksMoved);
         }
         return builder.obj();
     }
 
 private:
     const Timer _executionTimer;
-    Milliseconds _selectionTime;
-    Milliseconds _throttleTime;
-    Milliseconds _migrationTime;
 
     // Set only on success
-    int _numCandidateChunks{0};
-    int _numChunksMoved{0};
-    int _numImbalancedCachedCollections{0};
+    int _candidateChunks{0};
+    int _chunksMoved{0};
 
     // Set only on failure
     boost::optional<string> _errMsg;
 };
 
+/**
+ * Occasionally prints a log message with shard versions if the versions are not the same
+ * in the cluster.
+ */
+void warnOnMultiVersion(const vector<ClusterStatistics::ShardStatistics>& clusterStats) {
+    static const auto& majorMinorRE = *new pcrecpp::RE(R"re(^(\d+)\.(\d+)\.)re");
+    auto&& vii = VersionInfoInterface::instance();
+    auto hasMyVersion = [&](auto&& stat) {
+        int major;
+        int minor;
+        return majorMinorRE.PartialMatch(pcrecpp::StringPiece(stat.mongoVersion), &major, &minor) &&
+            major == vii.majorVersion() && minor == vii.minorVersion();
+    };
+
+    // If we're all the same version, don't message
+    if (std::all_of(clusterStats.begin(), clusterStats.end(), hasMyVersion))
+        return;
+
+    BSONObjBuilder shardVersions;
+    for (const auto& stat : clusterStats) {
+        shardVersions << stat.shardId << stat.mongoVersion;
+    }
+
+    LOGV2_WARNING(21875,
+                  "Multiversion cluster detected",
+                  "localVersion"_attr = vii.version(),
+                  "shardVersions"_attr = shardVersions.done());
+}
+
 Status processManualMigrationOutcome(OperationContext* opCtx,
                                      const BSONObj& chunkMin,
                                      const NamespaceString& nss,
@@ -215,136 +229,6 @@ const auto _balancerDecoration = ServiceContext::declareDecoration<Balancer>();
 
 const ReplicaSetAwareServiceRegistry::Registerer<Balancer> _balancerRegisterer("Balancer");
 
-/**
- * Returns the names of shards that are currently draining. When the balancer is disabled, draining
- * shards are stuck in this state as chunks cannot be migrated.
- */
-std::vector<std::string> getDrainingShardNames(OperationContext* opCtx) {
-    // Find the shards that are currently draining.
-    const auto configShard{Grid::get(opCtx)->shardRegistry()->getConfigShard()};
-    const auto drainingShardsDocs{
-        uassertStatusOK(
-            configShard->exhaustiveFindOnConfig(opCtx,
-                                                ReadPreferenceSetting{ReadPreference::Nearest},
-                                                repl::ReadConcernLevel::kMajorityReadConcern,
-                                                NamespaceString::kConfigsvrShardsNamespace,
-                                                BSON(ShardType::draining << true),
-                                                BSONObj() /* No sorting */,
-                                                boost::none /* No limit */))
-            .docs};
-
-    // Build the list of the draining shard names.
-    std::vector<std::string> drainingShardNames;
-    drainingShardNames.reserve(drainingShardsDocs.size());
-    std::transform(drainingShardsDocs.begin(),
-                   drainingShardsDocs.end(),
-                   std::back_inserter(drainingShardNames),
-                   [](const auto& shardDoc) {
-                       const auto shardEntry{uassertStatusOK(ShardType::fromBSON(shardDoc))};
-                       return shardEntry.getName();
-                   });
-    return drainingShardNames;
-}
-
-class BalancerWarning {
-    // Time interval between checks on draining shards.
-    constexpr static Minutes kDrainingShardsCheckInterval{10};
-
-public:
-    BalancerWarning() = default;
-
-    void warnIfRequired(OperationContext* opCtx, BalancerSettingsType::BalancerMode balancerMode) {
-        if (Date_t::now() - _lastDrainingShardsCheckTime < kDrainingShardsCheckInterval &&
-            MONGO_likely(!forceBalancerWarningChecks.shouldFail())) {
-            return;
-        }
-        _lastDrainingShardsCheckTime = Date_t::now();
-
-        LOGV2(7977401, "Performing balancer warning checks");
-
-        const auto drainingShardNames{getDrainingShardNames(opCtx)};
-        if (drainingShardNames.empty()) {
-            return;
-        }
-
-        if (balancerMode == BalancerSettingsType::BalancerMode::kOff) {
-            LOGV2_WARNING(
-                6434000,
-                "Draining of removed shards cannot be completed because the balancer is disabled",
-                "shards"_attr = drainingShardNames);
-            return;
-        }
-
-        _warnIfDrainingShardHasChunksForCollectionWithBalancingDisabled(opCtx, drainingShardNames);
-    }
-
-private:
-    void _warnIfDrainingShardHasChunksForCollectionWithBalancingDisabled(
-        OperationContext* opCtx, const std::vector<std::string>& drainingShardNames) {
-        // Balancer is on, emit warning if balancer is disabled for collections which have chunks in
-        // shards in draining mode.
-        const auto catalogClient = Grid::get(opCtx)->catalogClient();
-        auto collections =
-            catalogClient->getCollections(opCtx,
-                                          {},
-                                          repl::ReadConcernLevel::kMajorityReadConcern,
-                                          BSON(CollectionType::kNssFieldName << 1));
-        if (collections.empty()) {
-            return;
-        }
-
-        // Construct BSONArray of draining shard names.
-        const auto drainingShardNameArray = [&]() {
-            BSONArrayBuilder shardNameArrayBuilder;
-            std::for_each(drainingShardNames.begin(),
-                          drainingShardNames.end(),
-                          [&shardNameArrayBuilder](const auto& shardName) {
-                              shardNameArrayBuilder.append(shardName);
-                          });
-            return shardNameArrayBuilder.arr();
-        }();
-
-        // For each collection, check if the collection has balancing disabled. If it is disabled,
-        // checks if the collection has any chunks in any of the draining shards. In which case a
-        // warning is emitted.
-        for (const auto& collType : collections) {
-            if (!collType.getAllowBalance() || !collType.getAllowMigrations() ||
-                !collType.getPermitMigrations()) {
-                const auto findQuery =
-                    BSON(ChunkType::collectionUUID() << collType.getUuid() << ChunkType::shard()
-                                                     << BSON("$in" << drainingShardNameArray));
-
-                auto const configShard = Grid::get(opCtx)->shardRegistry()->getConfigShard();
-                auto findResponse = uassertStatusOK(configShard->exhaustiveFindOnConfig(
-                    opCtx,
-                    ReadPreferenceSetting(ReadPreference::PrimaryOnly),
-                    repl::ReadConcernLevel::kMajorityReadConcern,
-                    ChunkType::ConfigNS,
-                    findQuery,
-                    BSONObj(),
-                    boost::none));
-
-                const auto& chunks = findResponse.docs;
-                if (!chunks.empty()) {
-                    stdx::unordered_set<std::string> shardsWithChunks;
-                    std::for_each(
-                        chunks.begin(), chunks.end(), [&shardsWithChunks](const BSONObj& chunkObj) {
-                            shardsWithChunks.emplace(chunkObj.getStringField(ChunkType::shard()));
-                        });
-                    LOGV2_WARNING(
-                        7977400,
-                        "Draining of removed shards cannot be completed because the balancer is "
-                        "disabled for a collection which has chunks in those shards",
-                        "uuid"_attr = collType.getUuid(),
-                        "nss"_attr = collType.getNss(),
-                        "shardsWithChunks"_attr = shardsWithChunks);
-                }
-            }
-        }
-    }
-
-    Date_t _lastDrainingShardsCheckTime{Date_t::fromMillisSinceEpoch(0)};
-};
 }  // namespace
 
 Balancer* Balancer::get(ServiceContext* serviceContext) {
@@ -365,8 +249,7 @@ Balancer::Balancer()
       _defragmentationPolicy(std::make_unique<BalancerDefragmentationPolicyImpl>(
           _clusterStats.get(), _random, [this]() { _onActionsStreamPolicyStateUpdate(); })),
       _clusterChunksResizePolicy(std::make_unique<ClusterChunksResizePolicyImpl>(
-          [this] { _onActionsStreamPolicyStateUpdate(); })),
-      _imbalancedCollectionsCache(std::make_unique<stdx::unordered_set<NamespaceString>>()) {}
+          [this] { _onActionsStreamPolicyStateUpdate(); })) {}
 
 Balancer::~Balancer() {
     // Terminate the balancer thread so it doesn't leak memory.
@@ -397,7 +280,6 @@ void Balancer::onBecomeArbiter() {
 
 void Balancer::initiateBalancer(OperationContext* opCtx) {
     stdx::lock_guard<Latch> scopedLock(_mutex);
-    _imbalancedCollectionsCache->clear();
     invariant(_state == kStopped);
     _state = kRunning;
 
@@ -409,11 +291,11 @@ void Balancer::initiateBalancer(OperationContext* opCtx) {
 
 void Balancer::interruptBalancer() {
     stdx::lock_guard<Latch> scopedLock(_mutex);
-    if (_state != kRunning) {
+    if (_state != kRunning)
         return;
-    }
 
     _state = kStopping;
+    _thread.detach();
 
     // Interrupt the balancer thread if it has been started. We are guaranteed that the operation
     // context of that thread is still alive, because we hold the balancer mutex.
@@ -428,10 +310,8 @@ void Balancer::interruptBalancer() {
 
 void Balancer::waitForBalancerToStop() {
     stdx::unique_lock<Latch> scopedLock(_mutex);
+
     _joinCond.wait(scopedLock, [this] { return _state == kStopped; });
-    if (_thread.joinable()) {
-        _thread.join();
-    }
 }
 
 void Balancer::joinCurrentRound(OperationContext* opCtx) {
@@ -554,16 +434,15 @@ void Balancer::report(OperationContext* opCtx, BSONObjBuilder* builder) {
     builder->append("mode", BalancerSettingsType::kBalancerModes[mode]);
     builder->append("inBalancerRound", _inBalancerRound);
     builder->append("numBalancerRounds", _numBalancerRounds);
-    builder->append("term", repl::ReplicationCoordinator::get(opCtx)->getTerm());
 }
 
 void Balancer::_consumeActionStreamLoop() {
-    Client::initThread("BalancerSecondary");
-    {
-        stdx::lock_guard<Client> lk(cc());
-        cc().setSystemOperationKillableByStepdown(lk);
-    }
+    ScopeGuard onExitCleanup([this] {
+        _defragmentationPolicy->interruptAllDefragmentations();
+        _clusterChunksResizePolicy->stop();
+    });
 
+    Client::initThread("BalancerSecondary");
     auto opCtx = cc().makeOperationContext();
     // This thread never refreshes balancerConfig - instead, it relies on the requests
     // performed by _mainThread() on each round to eventually see updated information.
@@ -571,20 +450,6 @@ void Balancer::_consumeActionStreamLoop() {
     executor::ScopedTaskExecutor executor(
         Grid::get(opCtx.get())->getExecutorPool()->getFixedExecutor());
 
-    ScopeGuard onExitCleanup([this, &executor] {
-        _defragmentationPolicy->interruptAllDefragmentations();
-        _clusterChunksResizePolicy->stop();
-        // Explicitly cancel and drain any outstanding streaming action already dispatched to the
-        // task executor.
-        executor->shutdown();
-        executor->join();
-        // When shutting down, the task executor may or may not invoke the
-        // _applyDefragmentationActionResponseToPolicy() callback for canceled streaming actions: to
-        // ensure a consistent state of the balancer after a step down, _outstandingStreamingOps
-        // needs then to be reset to 0 once all the tasks have been drained.
-        _outstandingStreamingOps.store(0);
-    });
-
     auto selectStream = [&]() -> ActionsStreamPolicy* {
         // This policy has higher priority - and once activated, it cannot be disabled through cfg
         // changes.
@@ -598,6 +463,15 @@ void Balancer::_consumeActionStreamLoop() {
         return nullptr;
     };
 
+    auto applyActionResponseTo = [this](const DefragmentationAction& action,
+                                        const DefragmentationActionResponse& response,
+                                        ActionsStreamPolicy* policy) {
+        invariant(_outstandingStreamingOps.addAndFetch(-1) >= 0);
+        ThreadClient tc("BalancerSecondaryThread::applyActionResponse", getGlobalServiceContext());
+        auto opCtx = tc->makeOperationContext();
+        policy->applyActionResult(opCtx.get(), action, response);
+    };
+
     auto applyThrottling = [lastActionTime = Date_t::fromMillisSinceEpoch(0)]() mutable {
         const Milliseconds throttle{chunkDefragmentationThrottlingMS.load()};
         auto timeSinceLastAction = Date_t::now() - lastActionTime;
@@ -641,23 +515,9 @@ void Balancer::_consumeActionStreamLoop() {
                   "selectedStream"_attr = selectedStream->getName());
         }
 
-        boost::optional<DefragmentationAction> nextAction;
-        try {
-            _newInfoOnStreamingActions.store(false);
-            nextAction = selectedStream->getNextStreamingAction(opCtx.get());
-        } catch (const DBException& e) {
-            LOGV2_WARNING(7435001,
-                          "Failed to get next action from action stream",
-                          "error"_attr = redact(e),
-                          "stream"_attr = selectedStream->getName());
-
-            _newInfoOnStreamingActions.store(true);
-            continue;
-        }
-
-        if (!nextAction.is_initialized()) {
-            // No action was returned by this stream. This means that the stream is drained.
-            streamDrained = true;
+        _newInfoOnStreamingActions.store(false);
+        auto nextAction = selectedStream->getNextStreamingAction(opCtx.get());
+        if ((streamDrained = !nextAction.is_initialized())) {
             continue;
         }
 
@@ -674,10 +534,11 @@ void Balancer::_consumeActionStreamLoop() {
                                                  mergeAction.chunkRange,
                                                  mergeAction.collectionVersion)
                             .thenRunOn(*executor)
-                            .onCompletion([this, selectedStream, action = std::move(mergeAction)](
-                                              const Status& status) {
-                                _applyDefragmentationActionResponseToPolicy(
-                                    action, status, selectedStream);
+                            .onCompletion([this,
+                                           selectedStream,
+                                           &applyActionResponseTo,
+                                           action = std::move(mergeAction)](const Status& status) {
+                                applyActionResponseTo(action, status, selectedStream);
                             });
                 },
                 [&, selectedStream](DataSizeInfo&& dataSizeAction) {
@@ -689,15 +550,15 @@ void Balancer::_consumeActionStreamLoop() {
                                               dataSizeAction.chunkRange,
                                               dataSizeAction.version,
                                               dataSizeAction.keyPattern,
-                                              dataSizeAction.estimatedValue,
-                                              dataSizeAction.maxSize)
+                                              dataSizeAction.estimatedValue)
                             .thenRunOn(*executor)
-                            .onCompletion(
-                                [this, selectedStream, action = std::move(dataSizeAction)](
-                                    const StatusWith<DataSizeResponse>& swDataSize) {
-                                    _applyDefragmentationActionResponseToPolicy(
-                                        action, swDataSize, selectedStream);
-                                });
+                            .onCompletion([this,
+                                           selectedStream,
+                                           &applyActionResponseTo,
+                                           action = std::move(dataSizeAction)](
+                                              const StatusWith<DataSizeResponse>& swDataSize) {
+                                applyActionResponseTo(action, swDataSize, selectedStream);
+                            });
                 },
                 [&, selectedStream](AutoSplitVectorInfo&& splitVectorAction) {
                     auto result =
@@ -711,10 +572,12 @@ void Balancer::_consumeActionStreamLoop() {
                                                      splitVectorAction.maxChunkSizeBytes)
                             .thenRunOn(*executor)
                             .onCompletion(
-                                [this, selectedStream, action = std::move(splitVectorAction)](
+                                [this,
+                                 selectedStream,
+                                 &applyActionResponseTo,
+                                 action = std::move(splitVectorAction)](
                                     const StatusWith<AutoSplitVectorResponse>& swSplitPoints) {
-                                    _applyDefragmentationActionResponseToPolicy(
-                                        action, swSplitPoints, selectedStream);
+                                    applyActionResponseTo(action, swSplitPoints, selectedStream);
                                 });
                 },
                 [&, selectedStream](SplitInfoWithKeyPattern&& splitAction) {
@@ -730,10 +593,11 @@ void Balancer::_consumeActionStreamLoop() {
                                                 splitAction.info.maxKey,
                                                 splitAction.info.splitKeys)
                             .thenRunOn(*executor)
-                            .onCompletion([this, selectedStream, action = std::move(splitAction)](
-                                              const Status& status) {
-                                _applyDefragmentationActionResponseToPolicy(
-                                    action, status, selectedStream);
+                            .onCompletion([this,
+                                           selectedStream,
+                                           &applyActionResponseTo,
+                                           action = std::move(splitAction)](const Status& status) {
+                                applyActionResponseTo(action, status, selectedStream);
                             });
                 },
                 [](MigrateInfo&& _) {
@@ -746,12 +610,12 @@ void Balancer::_consumeActionStreamLoop() {
 
 void Balancer::_mainThread() {
     ON_BLOCK_EXIT([this] {
-        {
-            stdx::lock_guard<Latch> scopedLock(_mutex);
-            _state = kStopped;
-            LOGV2_DEBUG(21855, 1, "Balancer thread terminated");
-        }
+        stdx::lock_guard<Latch> scopedLock(_mutex);
+
+        _state = kStopped;
         _joinCond.notify_all();
+
+        LOGV2_DEBUG(21855, 1, "Balancer thread terminated");
     });
 
     Client::initThread("Balancer");
@@ -798,8 +662,6 @@ void Balancer::_mainThread() {
     LOGV2(6036606, "Balancer worker thread initialised. Entering main loop.");
 
     // Main balancer loop
-    auto lastMigrationTime = Date_t::fromMillisSinceEpoch(0);
-    BalancerWarning balancerWarning;
     while (!_stopRequested()) {
         BalanceRoundDetails roundDetails;
 
@@ -820,9 +682,6 @@ void Balancer::_mainThread() {
                 continue;
             }
 
-            // Warn before we skip the iteration due to balancing being disabled.
-            balancerWarning.warnIfRequired(opCtx.get(), balancerConfig->getBalancerMode());
-
             if (!balancerConfig->shouldBalance() || _stopRequested() ||
                 _clusterChunksResizePolicy->isActive()) {
                 LOGV2_DEBUG(21859, 1, "Skipping balancing round because balancing is disabled");
@@ -830,31 +689,33 @@ void Balancer::_mainThread() {
                 continue;
             }
 
-            boost::optional<Milliseconds> forcedBalancerRoundInterval(boost::none);
-            overrideBalanceRoundInterval.execute([&](const BSONObj& data) {
-                forcedBalancerRoundInterval = Milliseconds(data["intervalMs"].numberInt());
-                LOGV2(21864,
-                      "overrideBalanceRoundInterval: using customized balancing interval",
-                      "balancerInterval"_attr = *forcedBalancerRoundInterval);
-            });
-
             // The current configuration is allowing the balancer to perform operations.
             // Unblock the secondary thread if needed.
             _defragmentationCondVar.notify_all();
+            {
+                LOGV2_DEBUG(21860,
+                            1,
+                            "Start balancing round. waitForDelete: {waitForDelete}, "
+                            "secondaryThrottle: {secondaryThrottle}",
+                            "Start balancing round",
+                            "waitForDelete"_attr = balancerConfig->waitForDelete(),
+                            "secondaryThrottle"_attr =
+                                balancerConfig->getSecondaryThrottle().toBSON());
+
+                static Occasionally sampler;
+                if (sampler.tick()) {
+                    warnOnMultiVersion(uassertStatusOK(_clusterStats->getStats(opCtx.get())));
+                }
 
-            LOGV2_DEBUG(21860,
-                        1,
-                        "Start balancing round. waitForDelete: {waitForDelete}, "
-                        "secondaryThrottle: {secondaryThrottle}",
-                        "Start balancing round",
-                        "waitForDelete"_attr = balancerConfig->waitForDelete(),
-                        "secondaryThrottle"_attr = balancerConfig->getSecondaryThrottle().toBSON());
-
-            // Collect and apply up-to-date configuration values on the cluster collections.
-            _defragmentationPolicy->startCollectionDefragmentations(opCtx.get());
+                // Collect and apply up-to-date configuration values on the cluster collections.
+                {
+                    OperationContext* ctx = opCtx.get();
+                    auto allCollections = Grid::get(ctx)->catalogClient()->getCollections(ctx, {});
+                    for (const auto& coll : allCollections) {
+                        _defragmentationPolicy->startCollectionDefragmentation(ctx, coll);
+                    }
+                }
 
-            // Split chunk to match zones boundaries
-            {
                 Status status = _splitChunksIfNeeded(opCtx.get());
                 if (!status.isOK()) {
                     LOGV2_WARNING(21878,
@@ -864,83 +725,47 @@ void Balancer::_mainThread() {
                 } else {
                     LOGV2_DEBUG(21861, 1, "Done enforcing tag range boundaries.");
                 }
-            }
-
-            // Select and migrate chunks
-            {
-                Timer selectionTimer;
 
-                const std::vector<ClusterStatistics::ShardStatistics> shardStats =
-                    uassertStatusOK(_clusterStats->getStats(opCtx.get()));
-
-                stdx::unordered_set<ShardId> availableShards;
-                std::transform(
-                    shardStats.begin(),
-                    shardStats.end(),
-                    std::inserter(availableShards, availableShards.end()),
-                    [](const ClusterStatistics::ShardStatistics& shardStatistics) -> ShardId {
-                        return shardStatistics.shardId;
-                    });
+                stdx::unordered_set<ShardId> usedShards;
 
                 const auto chunksToDefragment =
-                    _defragmentationPolicy->selectChunksToMove(opCtx.get(), &availableShards);
+                    _defragmentationPolicy->selectChunksToMove(opCtx.get(), &usedShards);
 
                 const auto chunksToRebalance = uassertStatusOK(
-                    _chunkSelectionPolicy->selectChunksToMove(opCtx.get(),
-                                                              shardStats,
-                                                              &availableShards,
-                                                              _imbalancedCollectionsCache.get()));
-                const Milliseconds selectionTimeMillis{selectionTimer.millis()};
+                    _chunkSelectionPolicy->selectChunksToMove(opCtx.get(), &usedShards));
 
                 if (chunksToRebalance.empty() && chunksToDefragment.empty()) {
                     LOGV2_DEBUG(21862, 1, "No need to move any chunk");
                     _balancedLastTime = 0;
-                    LOGV2_DEBUG(21863, 1, "End balancing round");
-                    _endRound(opCtx.get(),
-                              forcedBalancerRoundInterval ? *forcedBalancerRoundInterval
-                                                          : kBalanceRoundDefaultInterval);
                 } else {
-
-                    // Sleep according to the migration throttling settings
-                    const auto throttleTimeMillis = [&] {
-                        const auto& minRoundinterval = forcedBalancerRoundInterval
-                            ? *forcedBalancerRoundInterval
-                            : Milliseconds(balancerMigrationsThrottlingMs.load());
-
-                        const auto timeSinceLastMigration = Date_t::now() - lastMigrationTime;
-                        if (timeSinceLastMigration < minRoundinterval) {
-                            return minRoundinterval - timeSinceLastMigration;
-                        }
-                        return Milliseconds::zero();
-                    }();
-                    _sleepFor(opCtx.get(), throttleTimeMillis);
-
-                    // Migrate chunks
-                    Timer migrationTimer;
                     _balancedLastTime =
                         _moveChunks(opCtx.get(), chunksToRebalance, chunksToDefragment);
-                    lastMigrationTime = Date_t::now();
-                    const Milliseconds migrationTimeMillis{migrationTimer.millis()};
 
-                    // Complete round
                     roundDetails.setSucceeded(
                         static_cast<int>(chunksToRebalance.size() + chunksToDefragment.size()),
-                        _balancedLastTime,
-                        _imbalancedCollectionsCache->size(),
-                        selectionTimeMillis,
-                        throttleTimeMillis,
-                        migrationTimeMillis);
+                        _balancedLastTime);
 
                     ShardingLogging::get(opCtx.get())
                         ->logAction(opCtx.get(), "balancer.round", "", roundDetails.toBSON())
                         .ignore();
-
-                    LOGV2_DEBUG(6679500, 1, "End balancing round");
-                    // Migration throttling of `balancerMigrationsThrottlingMs` will be applied
-                    // before the next call to _moveChunks, so don't sleep here.
-                    _endRound(opCtx.get(), Milliseconds(0));
                 }
+
+                LOGV2_DEBUG(21863, 1, "End balancing round");
             }
+
+            Milliseconds balancerInterval =
+                _balancedLastTime ? kShortBalanceRoundInterval : kBalanceRoundDefaultInterval;
+
+            overrideBalanceRoundInterval.execute([&](const BSONObj& data) {
+                balancerInterval = Milliseconds(data["intervalMs"].numberInt());
+                LOGV2(21864,
+                      "overrideBalanceRoundInterval: using shorter balancing interval: "
+                      "{balancerInterval}",
+                      "overrideBalanceRoundInterval: using shorter balancing interval",
+                      "balancerInterval"_attr = balancerInterval);
+            });
+
+            _endRound(opCtx.get(), balancerInterval);
         } catch (const DBException& e) {
             LOGV2(21865,
                   "caught exception while doing balance: {error}",
@@ -980,21 +805,6 @@ void Balancer::_mainThread() {
     LOGV2(21867, "CSRS balancer is now stopped");
 }
 
-void Balancer::_applyDefragmentationActionResponseToPolicy(
-    const DefragmentationAction& action,
-    const DefragmentationActionResponse& response,
-    ActionsStreamPolicy* policy) {
-    invariant(_outstandingStreamingOps.addAndFetch(-1) >= 0);
-    ThreadClient tc("BalancerSecondaryThread::applyActionResponse", getGlobalServiceContext());
-    {
-        stdx::lock_guard<Client> lk(cc());
-        cc().setSystemOperationKillableByStepdown(lk);
-    }
-
-    auto opCtx = tc->makeOperationContext();
-    policy->applyActionResult(opCtx.get(), action, response);
-};
-
 bool Balancer::_stopRequested() {
     stdx::lock_guard<Latch> scopedLock(_mutex);
     return (_state != kRunning);
@@ -1154,17 +964,13 @@ int Balancer::_moveChunks(OperationContext* opCtx,
     std::vector<std::pair<const MigrateInfo&, SemiFuture<void>>> rebalanceMigrationsAndResponses,
         defragmentationMigrationsAndResponses;
     auto requestMigration = [&](const MigrateInfo& migrateInfo) -> SemiFuture<void> {
-        auto maxChunkSizeBytes = [&]() {
-            if (migrateInfo.optMaxChunkSizeBytes.has_value()) {
-                return *migrateInfo.optMaxChunkSizeBytes;
-            }
-
-            auto coll = Grid::get(opCtx)->catalogClient()->getCollection(
-                opCtx, migrateInfo.nss, repl::ReadConcernLevel::kMajorityReadConcern);
-            return coll.getMaxChunkSizeBytes().value_or(balancerConfig->getMaxChunkSizeBytes());
-        }();
+        auto coll = Grid::get(opCtx)->catalogClient()->getCollection(
+            opCtx, migrateInfo.nss, repl::ReadConcernLevel::kMajorityReadConcern);
+        auto maxChunkSizeBytes =
+            coll.getMaxChunkSizeBytes().value_or(balancerConfig->getMaxChunkSizeBytes());
 
-        if (migrateInfo.maxKey.has_value()) {
+        if (serverGlobalParams.featureCompatibility.isLessThan(
+                multiversion::FeatureCompatibilityVersion::kVersion_6_0)) {
             // TODO SERVER-65322 only use `moveRange` once v6.0 branches out
             MoveChunkSettings settings(maxChunkSizeBytes,
                                        balancerConfig->getSecondaryThrottle(),
@@ -1175,24 +981,18 @@ int Balancer::_moveChunks(OperationContext* opCtx,
         MoveRangeRequestBase requestBase(migrateInfo.to);
         requestBase.setWaitForDelete(balancerConfig->waitForDelete());
         requestBase.setMin(migrateInfo.minKey);
-        requestBase.setMax(migrateInfo.maxKey);
+        if (!feature_flags::gNoMoreAutoSplitter.isEnabled(
+                serverGlobalParams.featureCompatibility)) {
+            // Issue the equivalent of a `moveChunk` if the auto-splitter is enabled
+            requestBase.setMax(migrateInfo.maxKey);
+        }
 
         ShardsvrMoveRange shardSvrRequest(migrateInfo.nss);
         shardSvrRequest.setDbName(NamespaceString::kAdminDb);
         shardSvrRequest.setMoveRangeRequestBase(requestBase);
         shardSvrRequest.setMaxChunkSizeBytes(maxChunkSizeBytes);
         shardSvrRequest.setFromShard(migrateInfo.from);
-        shardSvrRequest.setEpoch(migrateInfo.version.epoch());
-        const auto forceJumbo = [&]() {
-            if (migrateInfo.forceJumbo == MoveChunkRequest::ForceJumbo::kForceManual) {
-                return ForceJumbo::kForceManual;
-            }
-            if (migrateInfo.forceJumbo == MoveChunkRequest::ForceJumbo::kForceBalancer) {
-                return ForceJumbo::kForceBalancer;
-            }
-            return ForceJumbo::kDoNotForce;
-        }();
-        shardSvrRequest.setForceJumbo(forceJumbo);
+        shardSvrRequest.setEpoch(coll.getEpoch());
         const auto [secondaryThrottle, wc] =
             getSecondaryThrottleAndWriteConcern(balancerConfig->getSecondaryThrottle());
         shardSvrRequest.setSecondaryThrottle(secondaryThrottle);
@@ -1236,32 +1036,10 @@ int Balancer::_moveChunks(OperationContext* opCtx,
                 opCtx, migrateInfo.uuid, repl::ReadConcernLevel::kMajorityReadConcern);
 
             ShardingCatalogManager::get(opCtx)->splitOrMarkJumbo(
-                opCtx, collection.getNss(), migrateInfo.minKey, migrateInfo.getMaxChunkSizeBytes());
+                opCtx, collection.getNss(), migrateInfo.minKey);
             continue;
         }
 
-        if (status == ErrorCodes::IndexNotFound &&
-            gFeatureFlagShardKeyIndexOptionalHashedSharding.isEnabled(
-                serverGlobalParams.featureCompatibility)) {
-
-            const auto cm = uassertStatusOK(
-                Grid::get(opCtx)->catalogCache()->getCollectionRoutingInfoWithRefresh(
-                    opCtx, migrateInfo.nss));
-
-            if (cm.getShardKeyPattern().isHashedPattern()) {
-                LOGV2(78252,
-                      "Turning off balancing for hashed collection because migration failed due to "
-                      "missing shardkey index",
-                      "migrateInfo"_attr = redact(migrateInfo.toString()),
-                      "error"_attr = redact(status),
-                      "collection"_attr = migrateInfo.nss);
-
-                // Write to config.collections to turn off the balancer.
-                _disableBalancer(opCtx, migrateInfo.nss);
-                continue;
-            }
-        }
-
         LOGV2(21872,
               "Migration {migrateInfo} failed with {error}",
               "Migration failed",
@@ -1280,31 +1058,6 @@ int Balancer::_moveChunks(OperationContext* opCtx,
     return numChunksProcessed;
 }
 
-void Balancer::_disableBalancer(OperationContext* opCtx, NamespaceString nss) {
-    const auto configShard = Grid::get(opCtx)->shardRegistry()->getConfigShard();
-
-    BatchedCommandRequest updateRequest([&]() {
-        write_ops::UpdateCommandRequest updateOp(CollectionType::ConfigNS);
-        updateOp.setUpdates({[&] {
-            write_ops::UpdateOpEntry entry;
-            entry.setQ(BSON(CollectionType::kNssFieldName << nss.ns()));
-            entry.setU(write_ops::UpdateModification::parseFromClassicUpdate(
-                BSON("$set" << BSON("noBalance" << true))));
-            entry.setMulti(false);
-            entry.setUpsert(false);
-            return entry;
-        }()});
-        return updateOp;
-    }());
-
-    auto response = configShard->runBatchWriteCommand(opCtx,
-                                                      Shard::kDefaultConfigCommandTimeout,
-                                                      updateRequest,
-                                                      ShardingCatalogClient::kMajorityWriteConcern,
-                                                      Shard::RetryPolicy::kIdempotent);
-    uassertStatusOK(response.toStatus());
-}
-
 void Balancer::_onActionsStreamPolicyStateUpdate() {
     // On any internal update of the defragmentation/cluster chunks resize policy status,
     // wake up the thread consuming the stream of actions
@@ -1322,20 +1075,6 @@ void Balancer::abortCollectionDefragmentation(OperationContext* opCtx, const Nam
 
 SharedSemiFuture<void> Balancer::applyLegacyChunkSizeConstraintsOnClusterData(
     OperationContext* opCtx) {
-    // Remove the maxChunkSizeBytes from config.system.collections to make it compatible with
-    // the balancing strategy based on the number of collection chunks
-    try {
-        ShardingCatalogManager::get(opCtx)->configureCollectionBalancing(
-            opCtx,
-            NamespaceString::kLogicalSessionsNamespace,
-            0,
-            boost::none /*defragmentCollection*/,
-            boost::none /*enableAutoSplitter*/);
-    } catch (const ExceptionFor<ErrorCodes::NamespaceNotSharded>&) {
-        // config.system.collections does not appear in config.collections; continue.
-    }
-
-    // Ensure now that each collection in the cluster complies with its "maxChunkSize" constraint
     const auto balancerConfig = Grid::get(opCtx)->getBalancerConfiguration();
     uassertStatusOK(balancerConfig->refreshAndCheck(opCtx));
     auto futureOutcome =
@@ -1356,12 +1095,7 @@ BalancerCollectionStatusResponse Balancer::getBalancerStatusForNs(OperationConte
         uasserted(ErrorCodes::NamespaceNotSharded, "Collection unsharded or undefined");
     }
 
-
-    const auto maxChunkSizeBytes = getMaxChunkSizeBytes(opCtx, coll);
-    double maxChunkSizeMB = (double)maxChunkSizeBytes / (1024 * 1024);
-    // Keep only 2 decimal digits to return a readable value
-    maxChunkSizeMB = std::ceil(maxChunkSizeMB * 100.0) / 100.0;
-
+    const auto maxChunkSizeMB = getMaxChunkSizeMB(opCtx, coll);
     BalancerCollectionStatusResponse response(maxChunkSizeMB, true /*balancerCompliant*/);
     auto setViolationOnResponse = [&response](const StringData& reason,
                                               const boost::optional<BSONObj>& details =