SERVER-32198 Split CollectionShardingState::getMetadata into three methods

o getCurrentMetadataIfKnown - which returns the current filtering
metadata if any is available
o getMetadataForOperation - which returns the metadata which is required
by the current opertion, based on the OperationShardingState
o getCurrentMetadata - which returns the currently available filtering
metadata (or UNSHARDED if not known)

This is in preparation for making
getMetadataForOperation/getCurrentMetadata throw
StaleShardVersion exception if the metadata has not been loaded yet.

1 files changed, 6 insertions, 4 deletions

diff --git a/src/mongo/db/s/split_chunk.cpp b/src/mongo/db/s/split_chunk.cpp
index f82f8f56bf2..d3c0a6908db 100644
--- a/src/mongo/db/s/split_chunk.cpp
+++ b/src/mongo/db/s/split_chunk.cpp
@@ -96,16 +96,17 @@ bool checkIfSingleDoc(OperationContext* opCtx,
  */
 bool checkMetadataForSuccessfulSplitChunk(OperationContext* opCtx,
                                           const NamespaceString& nss,
+                                          const OID& epoch,
                                           const ChunkRange& chunkRange,
                                           const std::vector<BSONObj>& splitKeys) {
     const auto metadataAfterSplit = [&] {
         AutoGetCollection autoColl(opCtx, nss, MODE_IS);
-        return CollectionShardingState::get(opCtx, nss)->getMetadata(opCtx);
+        return CollectionShardingState::get(opCtx, nss)->getCurrentMetadata();
     }();
 
     uassert(ErrorCodes::StaleEpoch,
-            str::stream() << "Collection " << nss.ns() << " became unsharded",
-            metadataAfterSplit->isSharded());
+            str::stream() << "Collection " << nss.ns() << " changed since split start",
+            metadataAfterSplit->getCollVersion().epoch() == epoch);
 
     auto newChunkBounds(splitKeys);
     auto startKey = chunkRange.getMin();
@@ -208,7 +209,8 @@ StatusWith<boost::optional<ChunkRange>> splitChunk(OperationContext* opCtx,
     if (!commandStatus.isOK() || !writeConcernStatus.isOK()) {
         forceShardFilteringMetadataRefresh(opCtx, nss);
 
-        if (checkMetadataForSuccessfulSplitChunk(opCtx, nss, chunkRange, splitKeys)) {
+        if (checkMetadataForSuccessfulSplitChunk(
+                opCtx, nss, expectedCollectionEpoch, chunkRange, splitKeys)) {
             // Split was committed.
         } else if (!commandStatus.isOK()) {
             return commandStatus;