SERVER-41480 Increment major version on splits where the shard version equals the collection version

(cherry picked from commit 790609ffb6bc1dac0a3c13d1898d5884e21a1b7a)

7 files changed, 282 insertions, 83 deletions

diff --git a/jstests/sharding/major_version_check.js b/jstests/sharding/major_version_check.js
index eb6eccc1f1e..baa01aa7966 100644
--- a/jstests/sharding/major_version_check.js
+++ b/jstests/sharding/major_version_check.js
@@ -27,7 +27,7 @@ printjson(admin.runCommand({getShardVersion: coll + ""}));
 printjson(staleMongos.getDB("admin").runCommand({getShardVersion: coll + ""}));
 
 // Compare strings b/c timestamp comparison is a bit weird
-assert.eq(Timestamp(1, 2), admin.runCommand({getShardVersion: coll + ""}).version);
+assert.eq(Timestamp(2, 2), admin.runCommand({getShardVersion: coll + ""}).version);
 assert.eq(Timestamp(1, 0),
           staleMongos.getDB("admin").runCommand({getShardVersion: coll + ""}).version);
 
@@ -48,5 +48,21 @@ printjson(staleMongos.getDB("admin").runCommand({getShardVersion: coll + ""}));
 assert.eq(Timestamp(1, 0),
           staleMongos.getDB("admin").runCommand({getShardVersion: coll + ""}).version);
 
+// Run another split on the original chunk, which does not exist anymore (but the stale mongos
+// thinks it exists). This should fail and cause a refresh on the shard, updating its shard
+// version.
+assert.commandFailed(staleMongos.getDB("admin").runCommand(
+    {split: coll + "", bounds: [{_id: MinKey}, {_id: MaxKey}]}));
+
+// This findOne will cause a refresh on the router since the shard version has now been
+// increased.
+staleMongos.getCollection(coll + "").findOne();
+
+printjson(staleMongos.getDB("admin").runCommand({getShardVersion: coll + ""}));
+
+// The previously stale mongos should now be up-to-date.
+assert.eq(Timestamp(2, 2),
+          staleMongos.getDB("admin").runCommand({getShardVersion: coll + ""}).version);
+
 st.stop();
 })();
diff --git a/jstests/sharding/migration_failure.js b/jstests/sharding/migration_failure.js
index f731c0d3614..7f05dd1f9bb 100644
--- a/jstests/sharding/migration_failure.js
+++ b/jstests/sharding/migration_failure.js
@@ -30,6 +30,12 @@ var newVersion = null;
 assert.commandWorked(st.shard0.getDB("admin").runCommand(
     {configureFailPoint: 'failMigrationCommit', mode: 'alwaysOn'}));
 
+// The split command above bumps the shard version, and this is obtained by the router via a
+// refresh at the end of the command, but the shard does not know about it yet. This find will
+// cause the shard to refresh so that this next check for 'oldVersion' sees the most recent
+// version prior to the migration.
+coll.findOne();
+
 oldVersion = st.shard0.getDB("admin").runCommand({getShardVersion: coll.toString()}).global;
 
 assert.commandFailed(
diff --git a/jstests/sharding/zero_shard_version.js b/jstests/sharding/zero_shard_version.js
index 7d08bf34d36..209eff8e49c 100644
--- a/jstests/sharding/zero_shard_version.js
+++ b/jstests/sharding/zero_shard_version.js
@@ -87,7 +87,7 @@ testDB_s1.adminCommand({shardCollection: 'test.user', key: {x: 1}});
 testDB_s1.adminCommand({split: 'test.user', middle: {x: 0}});
 
 // shard0: 0|0|b,
-// shard1: 1|1|b, [-inf, 0), [0, inf)
+// shard1: 2|1|b, [-inf, 0), [0, inf)
 
 testDB_s1.user.insert({x: 1});
 testDB_s1.user.insert({x: -11});
@@ -97,45 +97,41 @@ assert.commandWorked(
 st.configRS.awaitLastOpCommitted();
 
 // Official config:
-// shard0: 2|0|b, [-inf, 0)
-// shard1: 2|1|b, [0, inf)
+// shard0: 3|0|b, [-inf, 0)
+// shard1: 3|1|b, [0, inf)
 //
 // Shard metadata:
 // shard0: 0|0|b
-// shard1: 2|1|b
+// shard1: 3|1|b
 //
 // mongos2: 2|0|a
 
 checkShardMajorVersion(st.rs0.getPrimary(), 0);
-checkShardMajorVersion(st.rs1.getPrimary(), 2);
+checkShardMajorVersion(st.rs1.getPrimary(), 3);
 
 // mongos2 still thinks that { x: 1 } belong to st.shard0.shardName, but should be able to
 // refresh it's metadata correctly.
 assert.neq(null, testDB_s2.user.findOne({x: 1}));
 
-checkShardMajorVersion(st.rs0.getPrimary(), 2);
-checkShardMajorVersion(st.rs1.getPrimary(), 2);
+checkShardMajorVersion(st.rs0.getPrimary(), 3);
+checkShardMajorVersion(st.rs1.getPrimary(), 3);
 
 // Set shard metadata to 2|0|b
 assert.neq(null, testDB_s2.user.findOne({x: -11}));
 
-checkShardMajorVersion(st.rs0.getPrimary(), 2);
-checkShardMajorVersion(st.rs1.getPrimary(), 2);
+checkShardMajorVersion(st.rs0.getPrimary(), 3);
+checkShardMajorVersion(st.rs1.getPrimary(), 3);
 
 // Official config:
-// shard0: 2|0|b, [-inf, 0)
-// shard1: 2|1|b, [0, inf)
+// shard0: 3|0|b, [-inf, 0)
+// shard1: 3|1|b, [0, inf)
 //
 // Shard metadata:
-// shard0: 2|0|b
-// shard1: 2|1|b
+// shard0: 3|0|b
+// shard1: 3|1|b
 //
 // mongos3: 2|0|a
 
-// 4th mongos still thinks that { x: 1 } belong to st.shard0.shardName, but should be able to
-// refresh it's metadata correctly.
-assert.neq(null, testDB_s3.user.findOne({x: 1}));
-
 ///////////////////////////////////////////////////////
 // Test mongos thinks unsharded when it's actually sharded
 // mongos current versions: s0: 0|0|0, s2, s3: 2|0|b
diff --git a/src/mongo/db/s/config/sharding_catalog_manager_chunk_operations.cpp b/src/mongo/db/s/config/sharding_catalog_manager_chunk_operations.cpp
index a1bd16dc8b8..0542d7f5843 100644
--- a/src/mongo/db/s/config/sharding_catalog_manager_chunk_operations.cpp
+++ b/src/mongo/db/s/config/sharding_catalog_manager_chunk_operations.cpp
@@ -264,6 +264,24 @@ boost::optional<ChunkType> getControlChunkForMigrate(OperationContext* opCtx,
     return uassertStatusOK(ChunkType::fromConfigBSON(response.docs.front()));
 }
 
+// Helper function to find collection version and shard version.
+StatusWith<ChunkVersion> getMaxChunkVersionFromQueryResponse(
+    const NamespaceString& nss, const StatusWith<Shard::QueryResponse>& queryResponse) {
+
+    if (!queryResponse.isOK()) {
+        return queryResponse.getStatus();
+    }
+
+    const auto& chunksVector = queryResponse.getValue().docs;
+    if (chunksVector.empty()) {
+        return {ErrorCodes::IllegalOperation,
+                str::stream() << "Collection '" << nss.ns()
+                              << "' no longer either exists, is sharded, or has chunks"};
+    }
+
+    return ChunkVersion::parseLegacyWithField(chunksVector.front(), ChunkType::lastmod());
+}
+
 }  // namespace
 
 Status ShardingCatalogManager::commitChunkSplit(OperationContext* opCtx,
@@ -278,42 +296,54 @@ Status ShardingCatalogManager::commitChunkSplit(OperationContext* opCtx,
     // move chunks on different collections to proceed in parallel
     Lock::ExclusiveLock lk(opCtx->lockState(), _kChunkOpLock);
 
-    std::string errmsg;
-
     // Get the max chunk version for this namespace.
-    auto findStatus = Grid::get(opCtx)->shardRegistry()->getConfigShard()->exhaustiveFindOnConfig(
-        opCtx,
-        ReadPreferenceSetting{ReadPreference::PrimaryOnly},
-        repl::ReadConcernLevel::kLocalReadConcern,
-        ChunkType::ConfigNS,
-        BSON("ns" << nss.ns()),
-        BSON(ChunkType::lastmod << -1),
-        1);
-
-    if (!findStatus.isOK()) {
-        return findStatus.getStatus();
-    }
+    auto swCollVersion = getMaxChunkVersionFromQueryResponse(
+        nss,
+        Grid::get(opCtx)->shardRegistry()->getConfigShard()->exhaustiveFindOnConfig(
+            opCtx,
+            ReadPreferenceSetting{ReadPreference::PrimaryOnly},
+            repl::ReadConcernLevel::kLocalReadConcern,
+            ChunkType::ConfigNS,
+            BSON("ns" << nss.ns()),          // Query all chunks for this namespace.
+            BSON(ChunkType::lastmod << -1),  // Sort by version.
+            1));                             // Limit 1.
 
-    const auto& chunksVector = findStatus.getValue().docs;
-    if (chunksVector.empty()) {
-        errmsg = str::stream() << "splitChunk cannot split chunk " << range.toString()
-                               << ". Collection '" << nss.ns()
-                               << "' no longer either exists, is sharded, or has chunks";
-        return {ErrorCodes::IllegalOperation, errmsg};
+    if (!swCollVersion.isOK()) {
+        return swCollVersion.getStatus().withContext(
+            str::stream() << "splitChunk cannot split chunk " << range.toString() << ".");
     }
 
-    ChunkVersion collVersion = uassertStatusOK(
-        ChunkVersion::parseLegacyWithField(chunksVector.front(), ChunkType::lastmod()));
+    auto collVersion = swCollVersion.getValue();
 
     // Return an error if collection epoch does not match epoch of request.
     if (collVersion.epoch() != requestEpoch) {
-        errmsg = str::stream() << "splitChunk cannot split chunk " << range.toString()
-                               << ". Collection '" << nss.ns() << "' was dropped and re-created."
-                               << " Current epoch: " << collVersion.epoch()
-                               << ", cmd epoch: " << requestEpoch;
-        return {ErrorCodes::StaleEpoch, errmsg};
+        return {ErrorCodes::StaleEpoch,
+                str::stream() << "splitChunk cannot split chunk " << range.toString()
+                              << ". Collection '" << nss.ns() << "' was dropped and re-created."
+                              << " Current epoch: " << collVersion.epoch()
+                              << ", cmd epoch: " << requestEpoch};
     }
 
+    // Get the shard version (max chunk version) for the shard requesting the split.
+    auto swShardVersion = getMaxChunkVersionFromQueryResponse(
+        nss,
+        Grid::get(opCtx)->shardRegistry()->getConfigShard()->exhaustiveFindOnConfig(
+            opCtx,
+            ReadPreferenceSetting{ReadPreference::PrimaryOnly},
+            repl::ReadConcernLevel::kLocalReadConcern,
+            ChunkType::ConfigNS,
+            BSON("ns" << nss.ns() << "shard"
+                      << shardName),         // Query all chunks for this namespace and shard.
+            BSON(ChunkType::lastmod << -1),  // Sort by version.
+            1));                             // Limit 1.
+
+    if (!swShardVersion.isOK()) {
+        return swShardVersion.getStatus().withContext(
+            str::stream() << "splitChunk cannot split chunk " << range.toString() << ".");
+    }
+
+    auto shardVersion = swShardVersion.getValue();
+
     // Find the chunk history.
     const auto origChunk = _findChunkOnConfig(opCtx, nss, range.getMin());
     if (!origChunk.isOK()) {
@@ -323,6 +353,11 @@ Status ShardingCatalogManager::commitChunkSplit(OperationContext* opCtx,
     std::vector<ChunkType> newChunks;
 
     ChunkVersion currentMaxVersion = collVersion;
+    // Increment the major version only if the shard that owns the chunk being split has version ==
+    // collection version. See SERVER-41480 for details.
+    if (shardVersion == collVersion) {
+        currentMaxVersion.incMajor();
+    }
 
     auto startKey = range.getMin();
     auto newChunkBounds(splitPoints);
@@ -489,27 +524,25 @@ Status ShardingCatalogManager::commitChunkMerge(OperationContext* opCtx,
     if (!validAfter) {
         return {ErrorCodes::IllegalOperation, "chunk operation requires validAfter timestamp"};
     }
-    // Get the chunk with the highest version for this namespace
-    auto findStatus = Grid::get(opCtx)->shardRegistry()->getConfigShard()->exhaustiveFindOnConfig(
-        opCtx,
-        ReadPreferenceSetting{ReadPreference::PrimaryOnly},
-        repl::ReadConcernLevel::kLocalReadConcern,
-        ChunkType::ConfigNS,
-        BSON("ns" << nss.ns()),
-        BSON(ChunkType::lastmod << -1),
-        1);
 
-    if (!findStatus.isOK()) {
-        return findStatus.getStatus();
-    }
+    // Get the max chunk version for this namespace.
+    auto swCollVersion = getMaxChunkVersionFromQueryResponse(
+        nss,
+        Grid::get(opCtx)->shardRegistry()->getConfigShard()->exhaustiveFindOnConfig(
+            opCtx,
+            ReadPreferenceSetting{ReadPreference::PrimaryOnly},
+            repl::ReadConcernLevel::kLocalReadConcern,
+            ChunkType::ConfigNS,
+            BSON("ns" << nss.ns()),          // Query all chunks for this namespace.
+            BSON(ChunkType::lastmod << -1),  // Sort by version.
+            1));                             // Limit 1.
 
-    const auto& chunksVector = findStatus.getValue().docs;
-    if (chunksVector.empty())
-        return {ErrorCodes::IllegalOperation,
-                "collection does not exist, isn't sharded, or has no chunks"};
+    if (!swCollVersion.isOK()) {
+        return swCollVersion.getStatus().withContext(str::stream()
+                                                     << "mergeChunk cannot merge chunks.");
+    }
 
-    ChunkVersion collVersion = uassertStatusOK(
-        ChunkVersion::parseLegacyWithField(chunksVector.front(), ChunkType::lastmod()));
+    auto collVersion = swCollVersion.getValue();
 
     // Return an error if epoch of chunk does not match epoch of request
     if (collVersion.epoch() != requestEpoch) {
diff --git a/src/mongo/db/s/config/sharding_catalog_manager_split_chunk_test.cpp b/src/mongo/db/s/config/sharding_catalog_manager_split_chunk_test.cpp
index 478a8374407..a2f0a293482 100644
--- a/src/mongo/db/s/config/sharding_catalog_manager_split_chunk_test.cpp
+++ b/src/mongo/db/s/config/sharding_catalog_manager_split_chunk_test.cpp
@@ -77,9 +77,9 @@ TEST_F(SplitChunkTest, SplitExistingChunkCorrectlyShouldSucceed) {
     auto chunkDoc = chunkDocStatus.getValue();
     ASSERT_BSONOBJ_EQ(chunkSplitPoint, chunkDoc.getMax());
 
-    // Check for increment on first chunkDoc's minor version
-    ASSERT_EQ(origVersion.majorVersion(), chunkDoc.getVersion().majorVersion());
-    ASSERT_EQ(origVersion.minorVersion() + 1, chunkDoc.getVersion().minorVersion());
+    // Check for increment on first chunkDoc's major version.
+    ASSERT_EQ(origVersion.majorVersion() + 1, chunkDoc.getVersion().majorVersion());
+    ASSERT_EQ(1u, chunkDoc.getVersion().minorVersion());
 
     // Make sure the history is there
     ASSERT_EQ(2UL, chunkDoc.getHistory().size());
@@ -91,9 +91,9 @@ TEST_F(SplitChunkTest, SplitExistingChunkCorrectlyShouldSucceed) {
     auto otherChunkDoc = otherChunkDocStatus.getValue();
     ASSERT_BSONOBJ_EQ(chunkMax, otherChunkDoc.getMax());
 
-    // Check for increment on second chunkDoc's minor version
-    ASSERT_EQ(origVersion.majorVersion(), otherChunkDoc.getVersion().majorVersion());
-    ASSERT_EQ(origVersion.minorVersion() + 2, otherChunkDoc.getVersion().minorVersion());
+    // Check for increment on second chunkDoc's minor version.
+    ASSERT_EQ(origVersion.majorVersion() + 1, otherChunkDoc.getVersion().majorVersion());
+    ASSERT_EQ(2u, otherChunkDoc.getVersion().minorVersion());
 
     // Make sure the history is there
     ASSERT_EQ(2UL, otherChunkDoc.getHistory().size());
@@ -138,9 +138,9 @@ TEST_F(SplitChunkTest, MultipleSplitsOnExistingChunkShouldSucceed) {
     auto chunkDoc = chunkDocStatus.getValue();
     ASSERT_BSONOBJ_EQ(chunkSplitPoint, chunkDoc.getMax());
 
-    // Check for increment on first chunkDoc's minor version
-    ASSERT_EQ(origVersion.majorVersion(), chunkDoc.getVersion().majorVersion());
-    ASSERT_EQ(origVersion.minorVersion() + 1, chunkDoc.getVersion().minorVersion());
+    // Check for increment on first chunkDoc's major version.
+    ASSERT_EQ(origVersion.majorVersion() + 1, chunkDoc.getVersion().majorVersion());
+    ASSERT_EQ(1u, chunkDoc.getVersion().minorVersion());
 
     // Make sure the history is there
     ASSERT_EQ(2UL, chunkDoc.getHistory().size());
@@ -152,9 +152,9 @@ TEST_F(SplitChunkTest, MultipleSplitsOnExistingChunkShouldSucceed) {
     auto midChunkDoc = midChunkDocStatus.getValue();
     ASSERT_BSONOBJ_EQ(chunkSplitPoint2, midChunkDoc.getMax());
 
-    // Check for increment on second chunkDoc's minor version
-    ASSERT_EQ(origVersion.majorVersion(), midChunkDoc.getVersion().majorVersion());
-    ASSERT_EQ(origVersion.minorVersion() + 2, midChunkDoc.getVersion().minorVersion());
+    // Check for increment on second chunkDoc's minor version.
+    ASSERT_EQ(origVersion.majorVersion() + 1, midChunkDoc.getVersion().majorVersion());
+    ASSERT_EQ(2u, midChunkDoc.getVersion().minorVersion());
 
     // Make sure the history is there
     ASSERT_EQ(2UL, midChunkDoc.getHistory().size());
@@ -166,9 +166,9 @@ TEST_F(SplitChunkTest, MultipleSplitsOnExistingChunkShouldSucceed) {
     auto lastChunkDoc = lastChunkDocStatus.getValue();
     ASSERT_BSONOBJ_EQ(chunkMax, lastChunkDoc.getMax());
 
-    // Check for increment on third chunkDoc's minor version
-    ASSERT_EQ(origVersion.majorVersion(), lastChunkDoc.getVersion().majorVersion());
-    ASSERT_EQ(origVersion.minorVersion() + 3, lastChunkDoc.getVersion().minorVersion());
+    // Check for increment on third chunkDoc's minor version.
+    ASSERT_EQ(origVersion.majorVersion() + 1, lastChunkDoc.getVersion().majorVersion());
+    ASSERT_EQ(3u, lastChunkDoc.getVersion().minorVersion());
 
     // Make sure the history is there
     ASSERT_EQ(2UL, lastChunkDoc.getHistory().size());
@@ -222,9 +222,12 @@ TEST_F(SplitChunkTest, NewSplitShouldClaimHighestVersion) {
     auto chunkDoc = chunkDocStatus.getValue();
     ASSERT_BSONOBJ_EQ(chunkSplitPoint, chunkDoc.getMax());
 
-    // Check for increment based on the competing chunk version
-    ASSERT_EQ(competingVersion.majorVersion(), chunkDoc.getVersion().majorVersion());
-    ASSERT_EQ(competingVersion.minorVersion() + 1, chunkDoc.getVersion().minorVersion());
+    // Check for major version increment based on the competing chunk version.
+    ASSERT_EQ(competingVersion.majorVersion() + 1, chunkDoc.getVersion().majorVersion());
+    // The minor version gets reset to 0 when the major version is incremented, and chunk splits
+    // increment the minor version after incrementing the major version, so we expect the minor
+    // version here to be 0 + 1 = 1.
+    ASSERT_EQ(1u, chunkDoc.getVersion().minorVersion());
 
     // Second chunkDoc should have range [chunkSplitPoint, chunkMax]
     auto otherChunkDocStatus = getChunkDoc(operationContext(), chunkSplitPoint);
@@ -233,9 +236,117 @@ TEST_F(SplitChunkTest, NewSplitShouldClaimHighestVersion) {
     auto otherChunkDoc = otherChunkDocStatus.getValue();
     ASSERT_BSONOBJ_EQ(chunkMax, otherChunkDoc.getMax());
 
-    // Check for increment based on the competing chunk version
+    // Check for increment based on the competing chunk version.
+    ASSERT_EQ(competingVersion.majorVersion() + 1, otherChunkDoc.getVersion().majorVersion());
+    // The minor version gets reset to 0 when the major version is incremented, and chunk splits
+    // increment the minor version after incrementing the major version for the first chunk in the
+    // split vector, so we expect the minor version here to be 0 + 1 + 1 = 2.
+    ASSERT_EQ(2u, otherChunkDoc.getVersion().minorVersion());
+}
+
+TEST_F(SplitChunkTest, SplitsOnShardWithLowerShardVersionDoesNotIncreaseCollectionVersion) {
+    ChunkType chunk, chunk2;
+    chunk.setNS(kNamespace);
+    chunk2.setNS(kNamespace);
+    auto collEpoch = OID::gen();
+
+    // Set up first chunk with lower version on shard0001. Its shard will not have shard version ==
+    // collection version, so splits to it should give it the collection version plus a minor
+    // version bump.
+    auto origVersion = ChunkVersion(1, 2, collEpoch);
+    chunk.setVersion(origVersion);
+    chunk.setShard(ShardId("shard0000"));
+    chunk.setMin(BSON("a" << 1));
+    chunk.setMax(BSON("a" << 10));
+
+    // Set up second chunk (chunk2) on shard0001. This has the higher version.
+    auto competingVersion = ChunkVersion(2, 1, collEpoch);
+    chunk2.setVersion(competingVersion);
+    chunk2.setShard(ShardId("shard0001"));
+    chunk2.setMin(BSON("a" << 10));
+    chunk2.setMax(BSON("a" << 20));
+
+    setupChunks({chunk, chunk2});
+
+    std::vector<BSONObj> splitPoints;
+    auto chunkSplitPoint = BSON("a" << 5);
+    splitPoints.push_back(chunkSplitPoint);
+
+    ASSERT_OK(ShardingCatalogManager::get(operationContext())
+                  ->commitChunkSplit(operationContext(),
+                                     kNamespace,
+                                     collEpoch,
+                                     ChunkRange(chunk.getMin(), chunk.getMax()),
+                                     splitPoints,
+                                     chunk.getShard().toString()));
+
+    // First chunkDoc should have range [chunk.getMin(), chunkSplitPoint]
+    auto chunkDocStatus = getChunkDoc(operationContext(), chunk.getMin());
+    auto chunkDoc = chunkDocStatus.getValue();
+
+    // Check for major version increment based on the competing chunk version.
+    ASSERT_EQ(competingVersion.majorVersion(), chunkDoc.getVersion().majorVersion());
+    ASSERT_EQ(competingVersion.minorVersion() + 1u, chunkDoc.getVersion().minorVersion());
+
+    // Second chunkDoc should have range [chunkSplitPoint, chunkMax]
+    auto otherChunkDocStatus = getChunkDoc(operationContext(), chunkSplitPoint);
+    auto otherChunkDoc = otherChunkDocStatus.getValue();
+    // Check for increment based on the competing chunk version.
     ASSERT_EQ(competingVersion.majorVersion(), otherChunkDoc.getVersion().majorVersion());
-    ASSERT_EQ(competingVersion.minorVersion() + 2, otherChunkDoc.getVersion().minorVersion());
+    ASSERT_EQ(competingVersion.minorVersion() + 2u, otherChunkDoc.getVersion().minorVersion());
+}
+
+TEST_F(SplitChunkTest, SplitsOnShardWithHighestShardVersionIncreasesCollectionVersion) {
+    ChunkType chunk, chunk2;
+    chunk.setNS(kNamespace);
+    chunk2.setNS(kNamespace);
+    auto collEpoch = OID::gen();
+
+    // Set up first chunk with lower version on shard0001. Its shard will not have shard version ==
+    // collection version.
+    auto origVersion = ChunkVersion(1, 2, collEpoch);
+    chunk.setVersion(origVersion);
+    chunk.setShard(ShardId("shard0000"));
+    chunk.setMin(BSON("a" << 1));
+    chunk.setMax(BSON("a" << 10));
+
+    // Set up second chunk (chunk2) on shard0001. This has the higher version, so its shard version
+    // == collection version. When we split it, its major version should increase.
+    auto competingVersion = ChunkVersion(2, 1, collEpoch);
+    chunk2.setVersion(competingVersion);
+    chunk2.setShard(ShardId("shard0001"));
+    chunk2.setMin(BSON("a" << 10));
+    chunk2.setMax(BSON("a" << 20));
+
+    setupChunks({chunk, chunk2});
+
+    std::vector<BSONObj> splitPoints;
+    // This will split the second chunk.
+    auto chunkSplitPoint = BSON("a" << 15);
+    splitPoints.push_back(chunkSplitPoint);
+
+    ASSERT_OK(ShardingCatalogManager::get(operationContext())
+                  ->commitChunkSplit(operationContext(),
+                                     kNamespace,
+                                     collEpoch,
+                                     ChunkRange(chunk2.getMin(), chunk2.getMax()),
+                                     splitPoints,
+                                     chunk2.getShard().toString()));
+
+    // First chunkDoc should have range [chunk2.getMin(), chunkSplitPoint]
+    auto chunkDocStatus = getChunkDoc(operationContext(), chunk2.getMin());
+    auto chunkDoc = chunkDocStatus.getValue();
+
+    // Check for major version increment based on the competing chunk version.
+    ASSERT_EQ(competingVersion.majorVersion() + 1u, chunkDoc.getVersion().majorVersion());
+    ASSERT_EQ(1u, chunkDoc.getVersion().minorVersion());
+
+    // Second chunkDoc should have range [chunkSplitPoint, chunk2.getMax()]
+    auto otherChunkDocStatus = getChunkDoc(operationContext(), chunkSplitPoint);
+    auto otherChunkDoc = otherChunkDocStatus.getValue();
+    // Check for increment based on the competing chunk version.
+    ASSERT_EQ(competingVersion.majorVersion() + 1u, otherChunkDoc.getVersion().majorVersion());
+    ASSERT_EQ(2u, otherChunkDoc.getVersion().minorVersion());
 }
 
 TEST_F(SplitChunkTest, PreConditionFailErrors) {
diff --git a/src/mongo/s/chunk_version.h b/src/mongo/s/chunk_version.h
index a0d7f09c1db..c655a85b5d1 100644
--- a/src/mongo/s/chunk_version.h
+++ b/src/mongo/s/chunk_version.h
@@ -31,6 +31,7 @@
 
 #include "mongo/base/status_with.h"
 #include "mongo/db/jsobj.h"
+#include "mongo/util/assert_util.h"
 
 namespace mongo {
 
@@ -124,10 +125,21 @@ public:
     }
 
     void incMajor() {
+        uassert(
+            31180,
+            "The chunk major version has reached its maximum value. Manual intervention will be "
+            "required before more chunk move, split, or merge operations are allowed.",
+            majorVersion() != std::numeric_limits<uint32_t>::max());
         _combined = static_cast<uint64_t>(majorVersion() + 1) << 32;
     }
 
     void incMinor() {
+        uassert(
+            31181,
+            "The chunk minor version has reached its maximum value. Manual intervention will be "
+            "required before more chunk split or merge operations are allowed.",
+            minorVersion() != std::numeric_limits<uint32_t>::max());
+
         _combined++;
     }
 
diff --git a/src/mongo/s/chunk_version_test.cpp b/src/mongo/s/chunk_version_test.cpp
index 7b51e5197ce..0e1fd283a48 100644
--- a/src/mongo/s/chunk_version_test.cpp
+++ b/src/mongo/s/chunk_version_test.cpp
@@ -152,5 +152,30 @@ TEST(ChunkVersionConstruction, CreateWithLargeValues) {
     ASSERT_EQ(epoch, version.epoch());
 }
 
+TEST(ChunkVersionManipulation, ThrowsErrorIfOverflowIsAttemptedForMajorVersion) {
+    const uint32_t minorVersion = 0;
+    const uint32_t majorVersion = std::numeric_limits<uint32_t>::max();
+    const auto epoch = OID::gen();
+
+    ChunkVersion version(majorVersion, minorVersion, epoch);
+    ASSERT_EQ(majorVersion, version.majorVersion());
+    ASSERT_EQ(minorVersion, version.minorVersion());
+    ASSERT_EQ(epoch, version.epoch());
+
+    ASSERT_THROWS_CODE(version.incMajor(), DBException, 31180);
+}
+
+TEST(ChunkVersionManipulation, ThrowsErrorIfOverflowIsAttemptedForMinorVersion) {
+    const uint32_t minorVersion = std::numeric_limits<uint32_t>::max();
+    const uint32_t majorVersion = 0;
+    const auto epoch = OID::gen();
+
+    ChunkVersion version(majorVersion, minorVersion, epoch);
+    ASSERT_EQ(majorVersion, version.majorVersion());
+    ASSERT_EQ(minorVersion, version.minorVersion());
+    ASSERT_EQ(epoch, version.epoch());
+
+    ASSERT_THROWS_CODE(version.incMinor(), DBException, 31181);
+}
 }  // namespace
 }  // namespace mongo