SERVER-29979 Move SplitChunk Logic from split_chunk_command to split_chunk

split_chunk_command currently contains all of the SplitChunk logic. This commit moves the logic
into a separate module.

8 files changed, 1168 insertions, 288 deletions

diff --git a/src/mongo/db/s/SConscript b/src/mongo/db/s/SConscript
index f541fe72a52..cb7679d9c99 100644
--- a/src/mongo/db/s/SConscript
+++ b/src/mongo/db/s/SConscript
@@ -199,6 +199,7 @@ env.Library(
         'balancer',
         'collection_metadata',
         'sharding',
+        'split_chunk',
         'split_vector',
     ],
 )
@@ -271,6 +272,37 @@ env.CppUnitTest(
 )
 
 env.Library(
+    target='split_chunk',
+    source=[
+        'split_chunk.cpp',
+    ],
+    LIBDEPS=[
+        'sharding',
+        '$BUILD_DIR/mongo/base',
+        '$BUILD_DIR/mongo/bson/util/bson_extract',
+        '$BUILD_DIR/mongo/db/clientcursor',
+        '$BUILD_DIR/mongo/db/db_raii',
+        '$BUILD_DIR/mongo/db/dbhelpers',
+        '$BUILD_DIR/mongo/db/exec/exec',
+        '$BUILD_DIR/mongo/db/pipeline/serveronly',
+        '$BUILD_DIR/mongo/db/query/query',
+        '$BUILD_DIR/mongo/db/rw_concern_d',
+        '$BUILD_DIR/mongo/s/catalog/sharding_catalog_manager',
+    ]
+)
+
+env.CppUnitTest(
+    target='split_chunk_test',
+    source=[
+        'split_chunk_test.cpp'
+    ],
+    LIBDEPS=[
+        'split_chunk',
+        '$BUILD_DIR/mongo/s/shard_server_test_fixture',
+    ]
+)
+
+env.Library(
     target='chunk_size_tracker',
     source=[
         'chunk_size_tracker.cpp',
diff --git a/src/mongo/db/s/sharding_state.cpp b/src/mongo/db/s/sharding_state.cpp
index d9273b5cfc1..693b9085684 100644
--- a/src/mongo/db/s/sharding_state.cpp
+++ b/src/mongo/db/s/sharding_state.cpp
@@ -131,6 +131,11 @@ bool ShardingState::enabled() const {
     return _getInitializationState() == InitializationState::kInitialized;
 }
 
+void ShardingState::setEnabledForTest(const std::string& shardName) {
+    _setInitializationState(InitializationState::kInitialized);
+    _shardName = shardName;
+}
+
 Status ShardingState::canAcceptShardedCommands() const {
     if (serverGlobalParams.clusterRole != ClusterRole::ShardServer) {
         return {ErrorCodes::NoShardingEnabled,
diff --git a/src/mongo/db/s/sharding_state.h b/src/mongo/db/s/sharding_state.h
index d41764c1d01..d5f53976b7f 100644
--- a/src/mongo/db/s/sharding_state.h
+++ b/src/mongo/db/s/sharding_state.h
@@ -96,6 +96,12 @@ public:
     bool enabled() const;
 
     /**
+     * Force-sets the initialization state to InitializationState::kInitialized, for testing
+     * purposes. Note that this function should ONLY be used for testing purposes.
+     */
+    void setEnabledForTest(const std::string& shardName);
+
+    /**
      * Returns Status::OK if the ShardingState is enabled; if not, returns an error describing
      * whether the ShardingState is just not yet initialized, or if this shard is not running with
      * --shardsvr at all.
diff --git a/src/mongo/db/s/split_chunk.cpp b/src/mongo/db/s/split_chunk.cpp
new file mode 100644
index 00000000000..0b0ad9286e3
--- /dev/null
+++ b/src/mongo/db/s/split_chunk.cpp
@@ -0,0 +1,341 @@
+/**
+ *    Copyright (C) 2017 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    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
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    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 GNU Affero General 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_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kSharding
+
+#include "mongo/platform/basic.h"
+
+#include "mongo/db/s/split_chunk.h"
+
+#include "mongo/base/status_with.h"
+#include "mongo/bson/util/bson_extract.h"
+#include "mongo/db/commands.h"
+#include "mongo/db/db_raii.h"
+#include "mongo/db/dbhelpers.h"
+#include "mongo/db/index/index_descriptor.h"
+#include "mongo/db/keypattern.h"
+#include "mongo/db/namespace_string.h"
+#include "mongo/db/query/internal_plans.h"
+#include "mongo/db/s/collection_metadata.h"
+#include "mongo/db/s/collection_sharding_state.h"
+#include "mongo/db/s/sharding_state.h"
+#include "mongo/s/catalog/type_chunk.h"
+#include "mongo/s/client/shard_registry.h"
+#include "mongo/s/grid.h"
+#include "mongo/s/request_types/split_chunk_request_type.h"
+#include "mongo/util/log.h"
+
+namespace mongo {
+namespace {
+
+const ReadPreferenceSetting kPrimaryOnlyReadPreference{ReadPreference::PrimaryOnly};
+
+bool checkIfSingleDoc(OperationContext* opCtx,
+                      Collection* collection,
+                      const IndexDescriptor* idx,
+                      const ChunkType* chunk) {
+    KeyPattern kp(idx->keyPattern());
+    BSONObj newmin = Helpers::toKeyFormat(kp.extendRangeBound(chunk->getMin(), false));
+    BSONObj newmax = Helpers::toKeyFormat(kp.extendRangeBound(chunk->getMax(), true));
+
+    auto exec = InternalPlanner::indexScan(opCtx,
+                                           collection,
+                                           idx,
+                                           newmin,
+                                           newmax,
+                                           BoundInclusion::kIncludeStartKeyOnly,
+                                           PlanExecutor::NO_YIELD);
+    // check if exactly one document found
+    PlanExecutor::ExecState state;
+    BSONObj obj;
+    if (PlanExecutor::ADVANCED == (state = exec->getNext(&obj, NULL))) {
+        if (PlanExecutor::IS_EOF == (state = exec->getNext(&obj, NULL))) {
+            return true;
+        }
+    }
+
+    // Non-yielding collection scans from InternalPlanner will never error.
+    invariant(PlanExecutor::ADVANCED == state || PlanExecutor::IS_EOF == state);
+
+    return false;
+}
+
+/**
+ * Checks the collection's metadata for a successful split on the specified chunkRange using the
+ * specified splitKeys. Returns false if the metadata's chunks don't match the new chunk
+ * boundaries exactly.
+ */
+bool checkMetadataForSuccessfulSplitChunk(OperationContext* opCtx,
+                                          const NamespaceString& nss,
+                                          const ChunkRange& chunkRange,
+                                          const std::vector<BSONObj>& splitKeys) {
+    ScopedCollectionMetadata metadataAfterSplit;
+    {
+        AutoGetCollection autoColl(opCtx, nss, MODE_IS);
+
+        // Get collection metadata
+        metadataAfterSplit = CollectionShardingState::get(opCtx, nss.ns())->getMetadata();
+    }
+
+    auto newChunkBounds(splitKeys);
+    auto startKey = chunkRange.getMin();
+    newChunkBounds.push_back(chunkRange.getMax());
+
+    ChunkType nextChunk;
+    for (const auto& endKey : newChunkBounds) {
+        // Check that all new chunks fit the new chunk boundaries
+        if (!metadataAfterSplit->getNextChunk(startKey, &nextChunk) ||
+            nextChunk.getMax().woCompare(endKey)) {
+            return false;
+        }
+
+        startKey = endKey;
+    }
+
+    return true;
+}
+
+}  // anonymous namespace
+
+StatusWith<boost::optional<ChunkRange>> splitChunk(OperationContext* opCtx,
+                                                   const NamespaceString& nss,
+                                                   const BSONObj& keyPatternObj,
+                                                   const ChunkRange& chunkRange,
+                                                   const std::vector<BSONObj>& splitKeys,
+                                                   const std::string& shardName,
+                                                   const OID& expectedCollectionEpoch) {
+
+    ShardingState* shardingState = ShardingState::get(opCtx);
+    std::string errmsg;
+
+    const BSONObj min = chunkRange.getMin();
+    const BSONObj max = chunkRange.getMax();
+
+    //
+    // Lock the collection's metadata and get highest version for the current shard
+    // TODO(SERVER-25086): Remove distLock acquisition from split chunk
+    //
+    const std::string whyMessage(
+        str::stream() << "splitting chunk [" << min << ", " << max << ") in " << nss.toString());
+    auto scopedDistLock = Grid::get(opCtx)->catalogClient()->getDistLockManager()->lock(
+        opCtx, nss.ns(), whyMessage, DistLockManager::kSingleLockAttemptTimeout);
+    if (!scopedDistLock.isOK()) {
+        errmsg = str::stream() << "could not acquire collection lock for " << nss.toString()
+                               << " to split chunk [" << redact(min) << "," << redact(max) << ") "
+                               << causedBy(redact(scopedDistLock.getStatus()));
+        warning() << errmsg;
+        return scopedDistLock.getStatus();
+    }
+
+    // Always check our version remotely
+    ChunkVersion shardVersion;
+    Status refreshStatus = shardingState->refreshMetadataNow(opCtx, nss, &shardVersion);
+
+    if (!refreshStatus.isOK()) {
+        errmsg = str::stream() << "splitChunk cannot split chunk "
+                               << "[" << redact(min) << "," << redact(max) << ") "
+                               << causedBy(redact(refreshStatus));
+
+        warning() << errmsg;
+        return refreshStatus;
+    }
+
+    if (shardVersion.majorVersion() == 0) {
+        // It makes no sense to split if our version is zero and we have no chunks
+        errmsg = str::stream() << "splitChunk cannot split chunk "
+                               << "[" << redact(min) << "," << redact(max) << ") "
+                               << " with zero shard version";
+
+        warning() << errmsg;
+        return {ErrorCodes::CannotSplit, errmsg};
+    }
+
+    // Even though the splitChunk command transmits a value in the operation's shardVersion
+    // field, this value does not actually contain the shard version, but the global collection
+    // version.
+    if (expectedCollectionEpoch != shardVersion.epoch()) {
+        std::string msg = str::stream() << "splitChunk cannot split chunk "
+                                        << "[" << redact(min) << "," << redact(max) << "), "
+                                        << "collection '" << nss.ns() << "' may have been dropped. "
+                                        << "current epoch: " << shardVersion.epoch()
+                                        << ", cmd epoch: " << expectedCollectionEpoch;
+        warning() << msg;
+        return {ErrorCodes::StaleEpoch, msg};
+    }
+
+    ScopedCollectionMetadata collMetadata;
+    {
+        AutoGetCollection autoColl(opCtx, nss, MODE_IS);
+
+        // Get collection metadata
+        collMetadata = CollectionShardingState::get(opCtx, nss.ns())->getMetadata();
+    }
+
+    // With nonzero shard version, we must have metadata
+    invariant(collMetadata);
+
+    KeyPattern shardKeyPattern(collMetadata->getKeyPattern());
+
+    // If the shard uses a hashed key, then we must make sure that the split point is of
+    // type NumberLong.
+    if (KeyPattern::isHashedKeyPattern(shardKeyPattern.toBSON())) {
+        for (BSONObj splitKey : splitKeys) {
+            BSONObjIterator it(splitKey);
+            while (it.more()) {
+                BSONElement splitKeyElement = it.next();
+                if (splitKeyElement.type() != NumberLong) {
+                    errmsg = str::stream() << "splitChunk cannot split chunk [" << redact(min)
+                                           << "," << redact(max) << "), split point "
+                                           << splitKeyElement.toString()
+                                           << " must be of type "
+                                              "NumberLong for hashed shard key patterns";
+                    warning() << errmsg;
+                    return {ErrorCodes::CannotSplit, errmsg};
+                }
+            }
+        }
+    }
+
+    ChunkVersion collVersion = collMetadata->getCollVersion();
+    // With nonzero shard version, we must have a coll version >= our shard version
+    invariant(collVersion >= shardVersion);
+
+    {
+        ChunkType chunkToMove;
+        chunkToMove.setMin(min);
+        chunkToMove.setMax(max);
+        uassertStatusOK(collMetadata->checkChunkIsValid(chunkToMove));
+    }
+
+    // Commit the split to the config server.
+    auto request =
+        SplitChunkRequest(nss, shardName, expectedCollectionEpoch, chunkRange, splitKeys);
+
+    auto configCmdObj =
+        request.toConfigCommandBSON(ShardingCatalogClient::kMajorityWriteConcern.toBSON());
+
+    auto cmdResponseStatus =
+        Grid::get(opCtx)->shardRegistry()->getConfigShard()->runCommandWithFixedRetryAttempts(
+            opCtx,
+            kPrimaryOnlyReadPreference,
+            "admin",
+            configCmdObj,
+            Shard::RetryPolicy::kIdempotent);
+
+    //
+    // Refresh chunk metadata regardless of whether or not the split succeeded
+    //
+    {
+        ChunkVersion unusedShardVersion;
+        refreshStatus = shardingState->refreshMetadataNow(opCtx, nss, &unusedShardVersion);
+
+        if (!refreshStatus.isOK()) {
+            errmsg = str::stream() << "failed to refresh metadata for split chunk [" << redact(min)
+                                   << "," << redact(max) << ") " << causedBy(redact(refreshStatus));
+
+            warning() << errmsg;
+            return refreshStatus;
+        }
+    }
+
+    // If we failed to get any response from the config server at all, despite retries, then we
+    // should just go ahead and fail the whole operation.
+    if (!cmdResponseStatus.isOK()) {
+        return cmdResponseStatus.getStatus();
+    }
+
+    // Check commandStatus and writeConcernStatus
+    auto commandStatus = cmdResponseStatus.getValue().commandStatus;
+    auto writeConcernStatus = cmdResponseStatus.getValue().writeConcernStatus;
+
+    // Send stale epoch if epoch of request did not match epoch of collection
+    if (commandStatus == ErrorCodes::StaleEpoch) {
+        std::string msg = str::stream() << "splitChunk cannot split chunk "
+                                        << "[" << redact(min) << "," << redact(max) << "), "
+                                        << "collection '" << nss.ns() << "' may have been dropped. "
+                                        << "current epoch: " << collVersion.epoch()
+                                        << ", cmd epoch: " << expectedCollectionEpoch;
+        warning() << msg;
+
+        return {commandStatus.code(), str::stream() << msg << redact(causedBy(commandStatus))};
+    }
+
+    //
+    // If _configsvrCommitChunkSplit returned an error, look at this shard's metadata to
+    // determine if the split actually did happen. This can happen if there's a network error
+    // getting the response from the first call to _configsvrCommitChunkSplit, but it actually
+    // succeeds, thus the automatic retry fails with a precondition violation, for example.
+    //
+    if ((!commandStatus.isOK() || !writeConcernStatus.isOK()) &&
+        checkMetadataForSuccessfulSplitChunk(opCtx, nss, chunkRange, splitKeys)) {
+
+        LOG(1) << "splitChunk [" << redact(min) << "," << redact(max)
+               << ") has already been committed.";
+    } else if (!commandStatus.isOK()) {
+        return commandStatus;
+    } else if (!writeConcernStatus.isOK()) {
+        return writeConcernStatus;
+    }
+
+    AutoGetCollection autoColl(opCtx, nss, MODE_IS);
+
+    Collection* const collection = autoColl.getCollection();
+    if (!collection) {
+        warning() << "will not perform top-chunk checking since " << nss.toString()
+                  << " does not exist after splitting";
+        return boost::optional<ChunkRange>(boost::none);
+    }
+
+    // Allow multiKey based on the invariant that shard keys must be single-valued. Therefore,
+    // any multi-key index prefixed by shard key cannot be multikey over the shard key fields.
+    IndexDescriptor* idx =
+        collection->getIndexCatalog()->findShardKeyPrefixedIndex(opCtx, keyPatternObj, false);
+    if (!idx) {
+        return boost::optional<ChunkRange>(boost::none);
+    }
+
+    auto backChunk = ChunkType();
+    backChunk.setMin(splitKeys.back());
+    backChunk.setMax(max);
+
+    auto frontChunk = ChunkType();
+    frontChunk.setMin(min);
+    frontChunk.setMax(splitKeys.front());
+
+    if (shardKeyPattern.globalMax().woCompare(backChunk.getMax()) == 0 &&
+        checkIfSingleDoc(opCtx, collection, idx, &backChunk)) {
+        return boost::optional<ChunkRange>(ChunkRange(backChunk.getMin(), backChunk.getMax()));
+    } else if (shardKeyPattern.globalMin().woCompare(frontChunk.getMin()) == 0 &&
+               checkIfSingleDoc(opCtx, collection, idx, &frontChunk)) {
+        return boost::optional<ChunkRange>(ChunkRange(frontChunk.getMin(), frontChunk.getMax()));
+    }
+
+    return boost::optional<ChunkRange>(boost::none);
+}
+
+}  // namespace mongo
diff --git a/src/mongo/db/s/split_chunk.h b/src/mongo/db/s/split_chunk.h
new file mode 100644
index 00000000000..d4cd263054b
--- /dev/null
+++ b/src/mongo/db/s/split_chunk.h
@@ -0,0 +1,64 @@
+/**
+ *    Copyright (C) 2017 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    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
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    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 GNU Affero General 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.
+ */
+
+#pragma once
+
+#include <boost/optional.hpp>
+#include <string>
+#include <vector>
+
+namespace mongo {
+
+class BSONObj;
+class ChunkRange;
+class KeyPattern;
+class NamespaceString;
+class OID;
+class OperationContext;
+template <typename T>
+class StatusWith;
+
+/**
+ * Attempts to split a chunk with the specified parameters. If the split fails, then the StatusWith
+ * object returned will contain a Status with an ErrorCode regarding the cause of failure. If the
+ * split succeeds, then the StatusWith object returned will contain Status::Ok().
+ *
+ * Additionally, splitChunk will attempt to perform top-chunk optimization. If top-chunk
+ * optimization is performed, then the function will also return a ChunkRange, which contains the
+ * range for the top chunk. Note that this ChunkRange is boost::optional, meaning that if top-chunk
+ * optimization is not performed, boost::none will be returned inside of the StatusWith instead.
+ */
+StatusWith<boost::optional<ChunkRange>> splitChunk(OperationContext* opCtx,
+                                                   const NamespaceString& nss,
+                                                   const BSONObj& keyPatternObj,
+                                                   const ChunkRange& chunkRange,
+                                                   const std::vector<BSONObj>& splitKeys,
+                                                   const std::string& shardName,
+                                                   const OID& expectedCollectionEpoch);
+
+}  // namespace mongo
diff --git a/src/mongo/db/s/split_chunk_command.cpp b/src/mongo/db/s/split_chunk_command.cpp
index cc3b5dba493..bd783fd6b84 100644
--- a/src/mongo/db/s/split_chunk_command.cpp
+++ b/src/mongo/db/s/split_chunk_command.cpp
@@ -38,19 +38,9 @@
 #include "mongo/db/auth/authorization_session.h"
 #include "mongo/db/auth/privilege.h"
 #include "mongo/db/commands.h"
-#include "mongo/db/db_raii.h"
-#include "mongo/db/dbhelpers.h"
-#include "mongo/db/index/index_descriptor.h"
-#include "mongo/db/query/internal_plans.h"
-#include "mongo/db/s/collection_metadata.h"
-#include "mongo/db/s/collection_sharding_state.h"
 #include "mongo/db/s/operation_sharding_state.h"
 #include "mongo/db/s/sharding_state.h"
-#include "mongo/s/catalog/type_chunk.h"
-#include "mongo/s/chunk_version.h"
-#include "mongo/s/client/shard_registry.h"
-#include "mongo/s/grid.h"
-#include "mongo/s/request_types/split_chunk_request_type.h"
+#include "mongo/db/s/split_chunk.h"
 #include "mongo/s/stale_exception.h"
 #include "mongo/util/log.h"
 #include "mongo/util/mongoutils/str.h"
@@ -63,73 +53,6 @@ using std::vector;
 
 namespace {
 
-const ReadPreferenceSetting kPrimaryOnlyReadPreference{ReadPreference::PrimaryOnly};
-
-bool checkIfSingleDoc(OperationContext* opCtx,
-                      Collection* collection,
-                      const IndexDescriptor* idx,
-                      const ChunkType* chunk) {
-    KeyPattern kp(idx->keyPattern());
-    BSONObj newmin = Helpers::toKeyFormat(kp.extendRangeBound(chunk->getMin(), false));
-    BSONObj newmax = Helpers::toKeyFormat(kp.extendRangeBound(chunk->getMax(), true));
-
-    auto exec = InternalPlanner::indexScan(opCtx,
-                                           collection,
-                                           idx,
-                                           newmin,
-                                           newmax,
-                                           BoundInclusion::kIncludeStartKeyOnly,
-                                           PlanExecutor::NO_YIELD);
-    // check if exactly one document found
-    PlanExecutor::ExecState state;
-    BSONObj obj;
-    if (PlanExecutor::ADVANCED == (state = exec->getNext(&obj, NULL))) {
-        if (PlanExecutor::IS_EOF == (state = exec->getNext(&obj, NULL))) {
-            return true;
-        }
-    }
-
-    // Non-yielding collection scans from InternalPlanner will never error.
-    invariant(PlanExecutor::ADVANCED == state || PlanExecutor::IS_EOF == state);
-
-    return false;
-}
-
-//
-// Checks the collection's metadata for a successful split on the specified chunkRange
-// using the specified splitPoints. Returns false if the metadata's chunks don't match
-// the new chunk boundaries exactly.
-//
-bool _checkMetadataForSuccess(OperationContext* opCtx,
-                              const NamespaceString& nss,
-                              const ChunkRange& chunkRange,
-                              const std::vector<BSONObj>& splitKeys) {
-    ScopedCollectionMetadata metadataAfterSplit;
-    {
-        AutoGetCollection autoColl(opCtx, nss, MODE_IS);
-
-        // Get collection metadata
-        metadataAfterSplit = CollectionShardingState::get(opCtx, nss.ns())->getMetadata();
-    }
-
-    auto newChunkBounds(splitKeys);
-    auto startKey = chunkRange.getMin();
-    newChunkBounds.push_back(chunkRange.getMax());
-
-    ChunkType nextChunk;
-    for (const auto& endKey : newChunkBounds) {
-        // Check that all new chunks fit the new chunk boundaries
-        if (!metadataAfterSplit->getNextChunk(startKey, &nextChunk) ||
-            nextChunk.getMax().woCompare(endKey)) {
-            return false;
-        }
-
-        startKey = endKey;
-    }
-
-    return true;
-}
-
 class SplitChunkCommand : public ErrmsgCommandDeprecated {
 public:
     SplitChunkCommand() : ErrmsgCommandDeprecated("splitChunk") {}
@@ -172,8 +95,8 @@ public:
                    const BSONObj& cmdObj,
                    std::string& errmsg,
                    BSONObjBuilder& result) override {
-        auto shardingState = ShardingState::get(opCtx);
-        uassertStatusOK(shardingState->canAcceptShardedCommands());
+
+        uassertStatusOK(ShardingState::get(opCtx)->canAcceptShardedCommands());
 
         //
         // Check whether parameters passed to splitChunk are sound
@@ -199,8 +122,6 @@ public:
         }
 
         auto chunkRange = uassertStatusOK(ChunkRange::fromBSON(cmdObj));
-        const BSONObj min = chunkRange.getMin();
-        const BSONObj max = chunkRange.getMax();
 
         string shardName;
         auto parseShardNameStatus = bsonExtractStringField(cmdObj, "from", &shardName);
@@ -209,59 +130,6 @@ public:
 
         log() << "received splitChunk request: " << redact(cmdObj);
 
-        //
-        // Lock the collection's metadata and get highest version for the current shard
-        // TODO(SERVER-25086): Remove distLock acquisition from split chunk
-        //
-        const string whyMessage(str::stream() << "splitting chunk [" << min << ", " << max
-                                              << ") in "
-                                              << nss.toString());
-        auto scopedDistLock = Grid::get(opCtx)->catalogClient()->getDistLockManager()->lock(
-            opCtx, nss.ns(), whyMessage, DistLockManager::kSingleLockAttemptTimeout);
-        if (!scopedDistLock.isOK()) {
-            errmsg = str::stream() << "could not acquire collection lock for " << nss.toString()
-                                   << " to split chunk [" << redact(min) << "," << redact(max)
-                                   << ") " << causedBy(redact(scopedDistLock.getStatus()));
-            warning() << errmsg;
-            return appendCommandStatus(result, scopedDistLock.getStatus());
-        }
-
-        // Always check our version remotely
-        ChunkVersion shardVersion;
-        Status refreshStatus = shardingState->refreshMetadataNow(opCtx, nss, &shardVersion);
-
-        if (!refreshStatus.isOK()) {
-            errmsg = str::stream() << "splitChunk cannot split chunk "
-                                   << "[" << redact(min) << "," << redact(max) << ") "
-                                   << causedBy(redact(refreshStatus));
-
-            warning() << errmsg;
-            return false;
-        }
-
-        if (shardVersion.majorVersion() == 0) {
-            // It makes no sense to split if our version is zero and we have no chunks
-            errmsg = str::stream() << "splitChunk cannot split chunk "
-                                   << "[" << redact(min) << "," << redact(max) << ") "
-                                   << " with zero shard version";
-
-            warning() << errmsg;
-            return false;
-        }
-
-        ScopedCollectionMetadata collMetadata;
-        {
-            AutoGetCollection autoColl(opCtx, nss, MODE_IS);
-
-            // Get collection metadata
-            collMetadata = CollectionShardingState::get(opCtx, nss.ns())->getMetadata();
-        }
-
-        // With nonzero shard version, we must have metadata
-        invariant(collMetadata);
-
-        KeyPattern shardKeyPattern(collMetadata->getKeyPattern());
-
         vector<BSONObj> splitKeys;
         {
             BSONElement splitKeysElem;
@@ -272,28 +140,9 @@ public:
                 errmsg = "need to provide the split points to chunk over";
                 return false;
             }
-
             BSONObjIterator it(splitKeysElem.Obj());
-
-            // If the shard uses a hashed key, then we must make sure that the split point is of
-            // type NumberLong.
-            if (KeyPattern::isHashedKeyPattern(shardKeyPattern.toBSON())) {
-                while (it.more()) {
-                    BSONObj splitKeyObj = it.next().Obj();
-                    BSONObjIterator keyIt(splitKeyObj);
-                    while (keyIt.more()) {
-                        BSONElement splitKey = keyIt.next();
-                        if (splitKey.type() != NumberLong) {
-                            errmsg = "split point must be of type NumberLong";
-                            return false;
-                        }
-                    }
-                    splitKeys.push_back(splitKeyObj.getOwned());
-                }
-            } else {
-                while (it.more()) {
-                    splitKeys.push_back(it.next().Obj().getOwned());
-                }
+            while (it.more()) {
+                splitKeys.push_back(it.next().Obj().getOwned());
             }
         }
 
@@ -311,141 +160,19 @@ public:
             expectedCollectionEpoch = oss.getShardVersion(nss).epoch();
         }
 
-        // Even though the splitChunk command transmits a value in the operation's shardVersion
-        // field, this value does not actually contain the shard version, but the global collection
-        // version.
-        if (expectedCollectionEpoch != shardVersion.epoch()) {
-            std::string msg = str::stream() << "splitChunk cannot split chunk "
-                                            << "[" << redact(min) << "," << redact(max) << "), "
-                                            << "collection '" << nss.ns()
-                                            << "' may have been dropped. "
-                                            << "current epoch: " << shardVersion.epoch()
-                                            << ", cmd epoch: " << expectedCollectionEpoch;
-            warning() << msg;
-            return appendCommandStatus(result, {ErrorCodes::StaleEpoch, msg});
-        }
-
-        ChunkVersion collVersion = collMetadata->getCollVersion();
-        // With nonzero shard version, we must have a coll version >= our shard version
-        invariant(collVersion >= shardVersion);
-
-        {
-            ChunkType chunkToMove;
-            chunkToMove.setMin(min);
-            chunkToMove.setMax(max);
-            uassertStatusOK(collMetadata->checkChunkIsValid(chunkToMove));
-        }
-
-        // Commit the split to the config server.
-        auto request =
-            SplitChunkRequest(nss, shardName, expectedCollectionEpoch, chunkRange, splitKeys);
-
-        auto configCmdObj =
-            request.toConfigCommandBSON(ShardingCatalogClient::kMajorityWriteConcern.toBSON());
-
-        auto cmdResponseStatus =
-            Grid::get(opCtx)->shardRegistry()->getConfigShard()->runCommandWithFixedRetryAttempts(
-                opCtx,
-                kPrimaryOnlyReadPreference,
-                "admin",
-                configCmdObj,
-                Shard::RetryPolicy::kIdempotent);
-
-        //
-        // Refresh chunk metadata regardless of whether or not the split succeeded
-        //
-        {
-            ChunkVersion unusedShardVersion;
-            refreshStatus = shardingState->refreshMetadataNow(opCtx, nss, &unusedShardVersion);
-
-            if (!refreshStatus.isOK()) {
-                errmsg = str::stream() << "failed to refresh metadata for split chunk ["
-                                       << redact(min) << "," << redact(max) << ") "
-                                       << causedBy(redact(refreshStatus));
-
-                warning() << errmsg;
-                return false;
-            }
-        }
-
-        // If we failed to get any response from the config server at all, despite retries, then we
-        // should just go ahead and fail the whole operation.
-        if (!cmdResponseStatus.isOK())
-            return appendCommandStatus(result, cmdResponseStatus.getStatus());
+        auto statusWithOptionalChunkRange = splitChunk(
+            opCtx, nss, keyPatternObj, chunkRange, splitKeys, shardName, expectedCollectionEpoch);
 
-        // Check commandStatus and writeConcernStatus
-        auto commandStatus = cmdResponseStatus.getValue().commandStatus;
-        auto writeConcernStatus = cmdResponseStatus.getValue().writeConcernStatus;
+        // If the split chunk returns something that is not Status::Ok(), then something failed.
+        uassertStatusOK(statusWithOptionalChunkRange.getStatus());
 
-        // Send stale epoch if epoch of request did not match epoch of collection
-        if (commandStatus == ErrorCodes::StaleEpoch) {
-            std::string msg = str::stream() << "splitChunk cannot split chunk "
-                                            << "[" << redact(min) << "," << redact(max) << "), "
-                                            << "collection '" << nss.ns()
-                                            << "' may have been dropped. "
-                                            << "current epoch: " << collVersion.epoch()
-                                            << ", cmd epoch: " << expectedCollectionEpoch;
-            warning() << msg;
-
-            return appendCommandStatus(
-                result,
-                {commandStatus.code(), str::stream() << msg << redact(causedBy(commandStatus))});
-        }
-
-        //
-        // If _configsvrCommitChunkSplit returned an error, look at this shard's metadata to
-        // determine if  the split actually did happen. This can happen if there's a network error
-        // getting the response from the first call to _configsvrCommitChunkSplit, but it actually
-        // succeeds, thus the automatic retry fails with a precondition violation, for example.
-        //
-        if ((!commandStatus.isOK() || !writeConcernStatus.isOK()) &&
-            _checkMetadataForSuccess(opCtx, nss, chunkRange, splitKeys)) {
-
-            LOG(1) << "splitChunk [" << redact(min) << "," << redact(max)
-                   << ") has already been committed.";
-        } else if (!commandStatus.isOK()) {
-            return appendCommandStatus(result, commandStatus);
-        } else if (!writeConcernStatus.isOK()) {
-            return appendCommandStatus(result, writeConcernStatus);
-        }
-
-        // Select chunk to move out for "top chunk optimization".
-
-        AutoGetCollection autoColl(opCtx, nss, MODE_IS);
-
-        Collection* const collection = autoColl.getCollection();
-        if (!collection) {
-            warning() << "will not perform top-chunk checking since " << nss.toString()
-                      << " does not exist after splitting";
-            return true;
-        }
-
-        // Allow multiKey based on the invariant that shard keys must be single-valued. Therefore,
-        // any multi-key index prefixed by shard key cannot be multikey over the shard key fields.
-        IndexDescriptor* idx =
-            collection->getIndexCatalog()->findShardKeyPrefixedIndex(opCtx, keyPatternObj, false);
-        if (!idx) {
-            return true;
-        }
-
-        auto backChunk = ChunkType();
-        backChunk.setMin(splitKeys.back());
-        backChunk.setMax(max);
-
-        auto frontChunk = ChunkType();
-        frontChunk.setMin(min);
-        frontChunk.setMax(splitKeys.front());
-
-        if (shardKeyPattern.globalMax().woCompare(backChunk.getMax()) == 0 &&
-            checkIfSingleDoc(opCtx, collection, idx, &backChunk)) {
+        // Otherwise, we want to check whether or not top-chunk optimization should be performed.
+        // If yes, then we should have a ChunkRange that was returned. Regardless of whether it
+        // should be performed, we will return true.
+        if (auto topChunk = statusWithOptionalChunkRange.getValue()) {
             result.append("shouldMigrate",
-                          BSON("min" << backChunk.getMin() << "max" << backChunk.getMax()));
-        } else if (shardKeyPattern.globalMin().woCompare(frontChunk.getMin()) == 0 &&
-                   checkIfSingleDoc(opCtx, collection, idx, &frontChunk)) {
-            result.append("shouldMigrate",
-                          BSON("min" << frontChunk.getMin() << "max" << frontChunk.getMax()));
+                          BSON("min" << topChunk->getMin() << "max" << topChunk->getMax()));
         }
-
         return true;
     }
 
diff --git a/src/mongo/db/s/split_chunk_test.cpp b/src/mongo/db/s/split_chunk_test.cpp
new file mode 100644
index 00000000000..a90fb86ea6a
--- /dev/null
+++ b/src/mongo/db/s/split_chunk_test.cpp
@@ -0,0 +1,705 @@
+/*    Copyright 2017 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    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
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    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 GNU Affero General 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_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kSharding
+
+#include "mongo/platform/basic.h"
+
+#include <boost/optional/optional_io.hpp>
+
+#include "mongo/db/s/split_chunk.h"
+
+#include "mongo/client/remote_command_targeter_mock.h"
+#include "mongo/db/json.h"
+#include "mongo/db/s/sharding_state.h"
+#include "mongo/db/server_options.h"
+#include "mongo/executor/network_interface_mock.h"
+#include "mongo/executor/remote_command_request.h"
+#include "mongo/executor/remote_command_response.h"
+#include "mongo/executor/task_executor.h"
+#include "mongo/s/catalog/dist_lock_manager_mock.h"
+#include "mongo/s/catalog/type_chunk.h"
+#include "mongo/s/catalog/type_collection.h"
+#include "mongo/s/catalog/type_database.h"
+#include "mongo/s/catalog/type_shard.h"
+#include "mongo/s/catalog_cache.h"
+#include "mongo/s/client/shard_registry.h"
+#include "mongo/s/grid.h"
+#include "mongo/s/shard_server_test_fixture.h"
+#include "mongo/s/write_ops/batched_command_request.h"
+#include "mongo/s/write_ops/batched_command_response.h"
+#include "mongo/unittest/unittest.h"
+#include "mongo/util/log.h"
+
+namespace mongo {
+namespace {
+
+using executor::RemoteCommandRequest;
+using executor::RemoteCommandResponse;
+
+class SplitChunkTest : public ShardServerTestFixture {
+public:
+    void setUp() override {
+
+        ShardServerTestFixture::setUp();
+
+        // Initialize the CatalogCache so that shard server metadata refreshes will work.
+        catalogCache()->initializeReplicaSetRole(true);
+
+        // Instantiate names.
+        _epoch = OID::gen();
+
+        _shardId = ShardId("shardId");
+        _nss = NamespaceString(StringData("dbName"), StringData("collName"));
+
+        // Set up the databases collection
+        _db.setName("dbName");
+        _db.setPrimary(_shardId.toString());
+        _db.setSharded(true);
+        ASSERT_OK(_db.validate());
+
+        // Set up the collections collection
+        _coll.setNs(_nss);
+        _coll.setEpoch(_epoch);
+        _coll.setUpdatedAt(Date_t::fromMillisSinceEpoch(ChunkVersion(1, 3, _epoch).toLong()));
+        _coll.setKeyPattern(BSON("_id" << 1));
+        _coll.setUnique(false);
+        ASSERT_OK(_coll.validate());
+
+        // Set up the shard
+        _shard.setName(_shardId.toString());
+        _shard.setHost("TestHost1");
+        ASSERT_OK(_shard.validate());
+
+        setUpChunkRanges();
+        setUpChunkVersions();
+    }
+
+    /**
+     * Tells the DistLockManagerMock instance to expect a lock, and logs the corresponding
+     * information.
+     */
+    void expectLock();
+
+    /**
+     * Returns the mock response that correspond with particular requests. For example, dbResponse
+     * returns a response for when a find-databases request occurs.
+     *
+     * commitChunkSplitResponse    : responds with { "ok" : 1 } or { "ok" : 0 }
+     * dbResponse    : responds with vector containing _db.toBSON()
+     * collResponse  : responds with vector containing _coll.toBSON()
+     * shardResponse : responds with vector containing _shard.toBSON()
+     * chunkResponse : responds with vector containing all every chunk.toConfigBSON()
+     * emptyResponse : responds with empty vector
+     */
+    void commitChunkSplitResponse(bool isOk);
+    void dbResponse();
+    void collResponse();
+    void shardResponse();
+    void chunkResponse();
+    void emptyResponse();
+
+protected:
+    /**
+     * Helper functions to return vectors of basic chunk ranges, chunk versions to
+     * be used by some of the tests.
+     */
+    void setUpChunkRanges();
+    void setUpChunkVersions();
+
+    OID _epoch;
+
+    NamespaceString _nss;
+    ShardId _shardId;
+
+    DatabaseType _db;
+    CollectionType _coll;
+    ShardType _shard;
+
+    std::vector<ChunkRange> _chunkRanges;
+    std::vector<ChunkVersion> _chunkVersions;
+};
+
+void SplitChunkTest::setUpChunkRanges() {
+    BSONObjBuilder minKeyBuilder;
+    BSONObjBuilder maxKeyBuilder;
+    minKeyBuilder.appendMinKey("foo");
+    maxKeyBuilder.appendMaxKey("foo");
+
+    const BSONObj key1 = minKeyBuilder.obj();
+    const BSONObj key2 = BSON("foo" << 0);
+    const BSONObj key3 = BSON("foo" << 1024);
+    const BSONObj key4 = maxKeyBuilder.obj();
+
+    _chunkRanges.push_back(ChunkRange(key1, key2));
+    _chunkRanges.push_back(ChunkRange(key2, key3));
+    _chunkRanges.push_back(ChunkRange(key3, key4));
+}
+
+void SplitChunkTest::setUpChunkVersions() {
+    _chunkVersions = {
+        ChunkVersion(1, 1, _epoch), ChunkVersion(1, 2, _epoch), ChunkVersion(1, 3, _epoch)};
+}
+
+void SplitChunkTest::expectLock() {
+    dynamic_cast<DistLockManagerMock*>(distLock())
+        ->expectLock(
+            [this](StringData name, StringData whyMessage, Milliseconds) {
+                LOG(0) << name;
+                LOG(0) << whyMessage;
+            },
+            Status::OK());
+}
+
+void SplitChunkTest::commitChunkSplitResponse(bool isOk) {
+    onCommand([&](const RemoteCommandRequest& request) {
+        return isOk ? BSON("ok" << 1) : BSON("ok" << 0);
+    });
+}
+
+void SplitChunkTest::dbResponse() {
+    onFindCommand(
+        [&](const RemoteCommandRequest& request) { return std::vector<BSONObj>{_db.toBSON()}; });
+}
+
+void SplitChunkTest::collResponse() {
+    onFindCommand(
+        [&](const RemoteCommandRequest& request) { return std::vector<BSONObj>{_coll.toBSON()}; });
+}
+
+void SplitChunkTest::shardResponse() {
+    onFindCommand(
+        [&](const RemoteCommandRequest& request) { return std::vector<BSONObj>{_shard.toBSON()}; });
+}
+
+void SplitChunkTest::chunkResponse() {
+    onFindCommand([&](const RemoteCommandRequest& request) {
+        std::vector<BSONObj> response;
+        for (unsigned long i = 0; i < _chunkRanges.size(); ++i) {
+            ChunkType chunk(_nss, _chunkRanges[i], _chunkVersions[i], _shardId);
+            response.push_back(chunk.toConfigBSON());
+        }
+        return response;
+    });
+}
+
+void SplitChunkTest::emptyResponse() {
+    onFindCommand([&](const RemoteCommandRequest& request) { return std::vector<BSONObj>(); });
+}
+
+TEST_F(SplitChunkTest, ValidHashedKeyPatternSplitKeys) {
+
+    BSONObj keyPatternObj = BSON("foo"
+                                 << "hashed");
+    _coll.setKeyPattern(BSON("_id"
+                             << "hashed"));
+
+    // Build a vector of valid split keys, which are values of NumberLong types.
+    std::vector<BSONObj> validSplitKeys;
+    for (long long i = 256; i <= 1024; i += 256) {
+        validSplitKeys.push_back(BSON("foo" << i));
+    }
+
+    // Force-set the sharding state to enabled with the _shardId, for testing purposes.
+    ShardingState::get(operationContext())->setEnabledForTest(_shardId.toString());
+
+    expectLock();
+
+    // Call the splitChunk function asynchronously on a different thread, so that we do not block,
+    // and so we can construct the mock responses to requests made by splitChunk below.
+    auto future = launchAsync([&] {
+        auto statusWithOptionalChunkRange = splitChunk(operationContext(),
+                                                       _nss,
+                                                       keyPatternObj,
+                                                       _chunkRanges[1],
+                                                       validSplitKeys,
+                                                       _shardId.toString(),
+                                                       _epoch);
+        ASSERT_OK(statusWithOptionalChunkRange.getStatus());
+    });
+
+    // Here, we mock responses to the requests made by the splitChunk operation. The requests first
+    // do a find on the databases, then a find on all collections in the database we are looking
+    // for. Next, filter by the specific collection, and find the relevant chunks and shards.
+    dbResponse();
+    collResponse();
+    collResponse();
+    chunkResponse();
+    shardResponse();
+
+    // Because we provided valid split points, the config server should respond with { "ok" : 1 }.
+    commitChunkSplitResponse(true);
+
+    // Finally, we find the original collection, and then find the relevant chunks.
+    collResponse();
+    chunkResponse();
+
+    future.timed_get(kFutureTimeout);
+}
+
+TEST_F(SplitChunkTest, InvalidHashedKeyPatternSplitKeys) {
+
+    BSONObj keyPatternObj = BSON("foo"
+                                 << "hashed");
+    _coll.setKeyPattern(BSON("_id"
+                             << "hashed"));
+
+    // Build a vector of valid split keys, which contains values that may not necessarily be able
+    // to be converted to NumberLong types.
+    std::vector<BSONObj> invalidSplitKeys{BSON("foo" << (long long)20),
+                                          BSON("foo" << 512),
+                                          BSON("foo"
+                                               << "hello"),
+                                          BSON("foo"
+                                               << ""),
+                                          BSON("foo" << 3.1415926535)};
+
+    // Force-set the sharding state to enabled with the _shardId, for testing purposes.
+    ShardingState::get(operationContext())->setEnabledForTest(_shardId.toString());
+
+    expectLock();
+
+    // Call the splitChunk function asynchronously on a different thread, so that we do not block,
+    // and so we can construct the mock responses to requests made by splitChunk below.
+    auto future = launchAsync([&] {
+        auto statusWithOptionalChunkRange = splitChunk(operationContext(),
+                                                       _nss,
+                                                       keyPatternObj,
+                                                       _chunkRanges[1],
+                                                       invalidSplitKeys,
+                                                       _shardId.toString(),
+                                                       _epoch);
+        ASSERT_EQUALS(ErrorCodes::CannotSplit, statusWithOptionalChunkRange.getStatus());
+    });
+
+    // Here, we mock responses to the requests made by the splitChunk operation. The requests first
+    // do a find on the databases, then a find on all collections in the database we are looking
+    // for. Next, filter by the specific collection, and find the relevant chunks and shards.
+    dbResponse();
+    collResponse();
+    collResponse();
+    chunkResponse();
+    shardResponse();
+
+    future.timed_get(kFutureTimeout);
+}
+
+TEST_F(SplitChunkTest, ValidRangeKeyPatternSplitKeys) {
+
+    BSONObj keyPatternObj = BSON("foo" << 1);
+
+    // Build a vector of valid split keys, which contains values that may not necessarily be able
+    // be converted to NumberLong types. However, this does not matter since we are not using a
+    // hashed shard key pattern.
+    std::vector<BSONObj> validSplitKeys{BSON("foo" << 20),
+                                        BSON("foo" << 512),
+                                        BSON("foo"
+                                             << "hello"),
+                                        BSON("foo"
+                                             << ""),
+                                        BSON("foo" << 3.1415926535)};
+
+    // Force-set the sharding state to enabled with the _shardId, for testing purposes.
+    ShardingState::get(operationContext())->setEnabledForTest(_shardId.toString());
+
+    expectLock();
+
+    // Call the splitChunk function asynchronously on a different thread, so that we do not block,
+    // and so we can construct the mock responses to requests made by splitChunk below.
+    auto future = launchAsync([&] {
+        auto statusWithOptionalChunkRange = splitChunk(operationContext(),
+                                                       _nss,
+                                                       keyPatternObj,
+                                                       _chunkRanges[1],
+                                                       validSplitKeys,
+                                                       _shardId.toString(),
+                                                       _epoch);
+        ASSERT_OK(statusWithOptionalChunkRange.getStatus());
+    });
+
+    // Here, we mock responses to the requests made by the splitChunk operation. The requests first
+    // do a find on the databases, then a find on all collections in the database we are looking
+    // for. Next, filter by the specific collection, and find the relevant chunks and shards.
+    dbResponse();
+    collResponse();
+    collResponse();
+    chunkResponse();
+    shardResponse();
+
+    // Because we provided valid split points, the config server should respond with { "ok" : 1 }.
+    commitChunkSplitResponse(true);
+
+    // Finally, we find the original collection, and then find the relevant chunks.
+    collResponse();
+    chunkResponse();
+
+    future.timed_get(kFutureTimeout);
+}
+
+TEST_F(SplitChunkTest, SplitChunkWithNoErrors) {
+
+    BSONObj keyPatternObj = BSON("foo" << 1);
+
+    // Build a vector of split keys. Note that we start at {"foo" : 256} and end at {"foo" : 768},
+    // neither of which are boundary points.
+    std::vector<BSONObj> splitKeys;
+    for (int i = 256; i < 1024; i += 256) {
+        splitKeys.push_back(BSON("foo" << i));
+    }
+
+    // Force-set the sharding state to enabled with the _shardId, for testing purposes.
+    ShardingState::get(operationContext())->setEnabledForTest(_shardId.toString());
+
+    expectLock();
+
+    // Call the splitChunk function asynchronously on a different thread, so that we do not block,
+    // and so we can construct the mock responses to requests made by splitChunk below.
+    auto future = launchAsync([&] {
+        auto statusWithOptionalChunkRange = splitChunk(operationContext(),
+                                                       _nss,
+                                                       keyPatternObj,
+                                                       _chunkRanges[1],
+                                                       splitKeys,
+                                                       _shardId.toString(),
+                                                       _epoch);
+        ASSERT_OK(statusWithOptionalChunkRange.getStatus());
+    });
+
+    // Here, we mock responses to the requests made by the splitChunk operation. The requests first
+    // do a find on the databases, then a find on all collections in the database we are looking
+    // for. Next, filter by the specific collection, and find the relevant chunks and shards.
+    dbResponse();
+    collResponse();
+    collResponse();
+    chunkResponse();
+    shardResponse();
+
+    // Mock an OK response to the request to the config server regarding the chunk split, but first
+    // check the request parameters.
+    onCommand([&](const RemoteCommandRequest& request) {
+        ASSERT_EQ(HostAndPort("dummy", 123), request.target);
+        ASSERT_BSONOBJ_EQ(request.cmdObj["min"].Obj(), BSON("foo" << 0));
+        ASSERT_BSONOBJ_EQ(request.cmdObj["max"].Obj(), BSON("foo" << 1024));
+
+        // Check that the split points in the request are the same as the split keys that were
+        // initially passed to the splitChunk function.
+        std::vector<BSONElement> splitPoints = request.cmdObj["splitPoints"].Array();
+        ASSERT_EQ(splitKeys.size(), splitPoints.size());
+        int i = 0;
+        for (auto e : splitPoints) {
+            ASSERT(e.Obj().woCompare(splitKeys[i]) == 0);
+            i++;
+        }
+
+        return BSON("ok" << 1);
+    });
+
+    // Finally, we find the original collection, and then find the relevant chunks.
+    collResponse();
+    chunkResponse();
+
+    future.timed_get(kFutureTimeout);
+}
+
+TEST_F(SplitChunkTest, AttemptSplitWithConfigsvrError) {
+
+    BSONObj keyPatternObj = BSON("foo" << 1);
+
+    // Build a vector of split keys. Note that we start at {"foo" : 0} and end at {"foo" : 1024},
+    // both of which are boundary points.
+    std::vector<BSONObj> splitKeys;
+    for (int i = 0; i <= 1024; i += 256) {
+        splitKeys.push_back(BSON("foo" << i));
+    }
+
+    // Force-set the sharding state to enabled with the _shardId, for testing purposes.
+    ShardingState::get(operationContext())->setEnabledForTest(_shardId.toString());
+
+    expectLock();
+
+    // Call the splitChunk function asynchronously on a different thread, so that we do not block,
+    // and so we can construct the mock responses to requests made by splitChunk below.
+    auto future = launchAsync([&] {
+        auto statusWithOptionalChunkRange = splitChunk(operationContext(),
+                                                       _nss,
+                                                       keyPatternObj,
+                                                       _chunkRanges[1],
+                                                       splitKeys,
+                                                       _shardId.toString(),
+                                                       _epoch);
+        ASSERT_NOT_OK(statusWithOptionalChunkRange.getStatus());
+    });
+
+    // Here, we mock responses to the requests made by the splitChunk operation. The requests first
+    // do a find on the databases, then a find on all collections in the database we are looking
+    // for. Next, filter by the specific collection, and find the relevant chunks and shards.
+    dbResponse();
+    collResponse();
+    collResponse();
+    chunkResponse();
+    shardResponse();
+
+    // Because we provided invalid split points, the config server should respond with { "ok" : 0 }.
+    commitChunkSplitResponse(false);
+
+    // Finally, we find the original collection, and then find the relevant chunks.
+    collResponse();
+    chunkResponse();
+
+    future.timed_get(kFutureTimeout);
+}
+
+TEST_F(SplitChunkTest, AttemptSplitOnNoDatabases) {
+
+    BSONObj keyPatternObj = BSON("foo" << 1);
+
+    // Build a vector of split keys. Note that we start at {"foo" : 256} and end at {"foo" : 768},
+    // neither of which are boundary points.
+    std::vector<BSONObj> splitKeys;
+    for (int i = 256; i < 1024; i += 256) {
+        splitKeys.push_back(BSON("foo" << i));
+    }
+
+    // Force-set the sharding state to enabled with the _shardId, for testing purposes.
+    ShardingState::get(operationContext())->setEnabledForTest(_shardId.toString());
+
+    expectLock();
+
+    // Call the splitChunk function asynchronously on a different thread, so that we do not block,
+    // and so we can construct the mock responses to requests made by splitChunk below.
+    auto future = launchAsync([&] {
+        auto statusWithOptionalChunkRange = splitChunk(operationContext(),
+                                                       _nss,
+                                                       keyPatternObj,
+                                                       _chunkRanges[1],
+                                                       splitKeys,
+                                                       _shardId.toString(),
+                                                       _epoch);
+        ASSERT_NOT_OK(statusWithOptionalChunkRange.getStatus());
+    });
+
+    // Here, we mock responses to the requests made by the splitChunk operation. We give an empty
+    // response to the request that finds the database, along with the request that finds all
+    // collections in the database.
+    emptyResponse();
+    emptyResponse();
+
+    future.timed_get(kFutureTimeout);
+}
+
+TEST_F(SplitChunkTest, AttemptSplitOnNoCollections) {
+
+    BSONObj keyPatternObj = BSON("foo" << 1);
+
+    // Build a vector of split keys. Note that we start at {"foo" : 256} and end at {"foo" : 768},
+    // neither of which are boundary points.
+    std::vector<BSONObj> splitKeys;
+    for (int i = 256; i < 1024; i += 256) {
+        splitKeys.push_back(BSON("foo" << i));
+    }
+
+    // Force-set the sharding state to enabled with the _shardId, for testing purposes.
+    ShardingState::get(operationContext())->setEnabledForTest(_shardId.toString());
+
+    expectLock();
+
+    // Call the splitChunk function asynchronously on a different thread, so that we do not block,
+    // and so we can construct the mock responses to requests made by splitChunk below.
+    auto future = launchAsync([&] {
+        auto statusWithOptionalChunkRange = splitChunk(operationContext(),
+                                                       _nss,
+                                                       keyPatternObj,
+                                                       _chunkRanges[1],
+                                                       splitKeys,
+                                                       _shardId.toString(),
+                                                       _epoch);
+        ASSERT_NOT_OK(statusWithOptionalChunkRange.getStatus());
+    });
+
+    // Here, we mock responses to the requests made by the splitChunk operation. We first respond
+    // to the request finding the databases.
+    dbResponse();
+
+    // Next, we give an empty response to the request for finding collections in the database,
+    // followed by a response to the request for relevant shards.
+    emptyResponse();
+    shardResponse();
+
+    future.timed_get(kFutureTimeout);
+}
+
+TEST_F(SplitChunkTest, AttemptSplitOnNoChunks) {
+
+    BSONObj keyPatternObj = BSON("foo" << 1);
+
+    // Build a vector of split keys. Note that we start at {"foo" : 256} and end at {"foo" : 768},
+    // neither of which are boundary points.
+    std::vector<BSONObj> splitKeys;
+    for (int i = 256; i < 1024; i += 256) {
+        splitKeys.push_back(BSON("foo" << i));
+    }
+
+    // Force-set the sharding state to enabled with the _shardId, for testing purposes.
+    ShardingState::get(operationContext())->setEnabledForTest(_shardId.toString());
+
+    expectLock();
+
+    // Call the splitChunk function asynchronously on a different thread, so that we do not block,
+    // and so we can construct the mock responses to requests made by splitChunk below.
+    auto future = launchAsync([&] {
+        auto statusWithOptionalChunkRange = splitChunk(operationContext(),
+                                                       _nss,
+                                                       keyPatternObj,
+                                                       _chunkRanges[1],
+                                                       splitKeys,
+                                                       _shardId.toString(),
+                                                       _epoch);
+        ASSERT_NOT_OK(statusWithOptionalChunkRange.getStatus());
+    });
+
+    // Here, we mock responses to the requests made by the splitChunk operation. The requests first
+    // do a find on the databases, then a find on all collections in the database we are looking
+    // for.
+    dbResponse();
+    collResponse();
+
+    // We attempt to find the relevant chunks three times. For each of these times, we will respond
+    // with the relevant collection, but no chunks.
+    collResponse();
+    emptyResponse();
+    collResponse();
+    emptyResponse();
+    collResponse();
+    emptyResponse();
+
+    future.timed_get(kFutureTimeout);
+}
+
+TEST_F(SplitChunkTest, NoCollectionAfterSplit) {
+
+    BSONObj keyPatternObj = BSON("foo" << 1);
+
+    // Build a vector of split keys. Note that we start at {"foo" : 256} and end at {"foo" : 768},
+    // neither of which are boundary points.
+    std::vector<BSONObj> splitKeys;
+    for (int i = 256; i < 1024; i += 256) {
+        splitKeys.push_back(BSON("foo" << i));
+    }
+
+    // Force-set the sharding state to enabled with the _shardId, for testing purposes.
+    ShardingState::get(operationContext())->setEnabledForTest(_shardId.toString());
+
+    expectLock();
+
+    // Call the splitChunk function asynchronously on a different thread, so that we do not block,
+    // and so we can construct the mock responses to requests made by splitChunk below.
+    auto future = launchAsync([&] {
+        auto statusWithOptionalChunkRange = splitChunk(operationContext(),
+                                                       _nss,
+                                                       keyPatternObj,
+                                                       _chunkRanges[1],
+                                                       splitKeys,
+                                                       _shardId.toString(),
+                                                       _epoch);
+        ASSERT_OK(statusWithOptionalChunkRange.getStatus());
+    });
+
+    // Here, we mock responses to the requests made by the splitChunk operation. The requests first
+    // do a find on the databases, then a find on all collections in the database we are looking
+    // for. Next, filter by the specific collection, and find the relevant chunks and shards.
+    dbResponse();
+    collResponse();
+    collResponse();
+    chunkResponse();
+    shardResponse();
+
+    // Here, we mock a successful response from the config server, which denotes that the split was
+    // successful on the config server's end.
+    commitChunkSplitResponse(true);
+
+    // Finally, give an empty response to a request regarding a find on the original collection.
+    emptyResponse();
+
+    future.timed_get(kFutureTimeout);
+}
+
+TEST_F(SplitChunkTest, NoChunksAfterSplit) {
+
+    BSONObj keyPatternObj = BSON("foo" << 1);
+
+    // Build a vector of split keys. Note that we start at {"foo" : 256} and end at {"foo" : 768},
+    // neither of which are boundary points.
+    std::vector<BSONObj> splitKeys;
+    for (int i = 256; i < 1024; i += 256) {
+        splitKeys.push_back(BSON("foo" << i));
+    }
+
+    // Force-set the sharding state to enabled with the _shardId, for testing purposes.
+    ShardingState::get(operationContext())->setEnabledForTest(_shardId.toString());
+
+    expectLock();
+
+    // Call the splitChunk function asynchronously on a different thread, so that we do not block,
+    // and so we can construct the mock responses to requests made by splitChunk below.
+    auto future = launchAsync([&] {
+        auto statusWithOptionalChunkRange = splitChunk(operationContext(),
+                                                       _nss,
+                                                       keyPatternObj,
+                                                       _chunkRanges[1],
+                                                       splitKeys,
+                                                       _shardId.toString(),
+                                                       _epoch);
+        ASSERT_NOT_OK(statusWithOptionalChunkRange.getStatus());
+    });
+
+    // Here, we mock responses to the requests made by the splitChunk operation. The requests first
+    // do a find on the databases, then a find on all collections in the database we are looking
+    // for. Next, filter by the specific collection, and find the relevant chunks and shards.
+    dbResponse();
+    collResponse();
+    collResponse();
+    chunkResponse();
+    shardResponse();
+
+    // Here, we mock a successful response from the config server, which denotes that the split was
+    // successful on the config server's end.
+    commitChunkSplitResponse(true);
+
+    // We attempt to find the relevant chunks three times. For each of these times, we will respond
+    // with the relevant collection, but no chunks.
+    collResponse();
+    emptyResponse();
+    collResponse();
+    emptyResponse();
+    collResponse();
+    emptyResponse();
+
+    future.timed_get(kFutureTimeout);
+}
+
+}  // namespace
+}  // namespace mongo
diff --git a/src/mongo/s/shard_id.h b/src/mongo/s/shard_id.h
index 6456fc41677..76a31769453 100644
--- a/src/mongo/s/shard_id.h
+++ b/src/mongo/s/shard_id.h
@@ -81,7 +81,7 @@ public:
     int compare(const ShardId& other) const;
 
     /**
-     *  Returns true if _shardId is empty. Subject to include more validations in the future.
+     *  Returns true if _shardId is not empty. Subject to include more validations in the future.
      */
     bool isValid() const;