SERVER-42617 Race in CloneDocumentsCatchesInsertErrors can causes it to return an unexpected error

When the batch insertion routine on the consumer thread fails,
ensure that we return the proper error instead of
ErrorCodes::ProducerConsumerQueueEndClosed.

1 files changed, 20 insertions, 16 deletions

diff --git a/src/mongo/db/s/migration_destination_manager.cpp b/src/mongo/db/s/migration_destination_manager.cpp
index 54f4a01e87b..8124f36a16f 100644
--- a/src/mongo/db/s/migration_destination_manager.cpp
+++ b/src/mongo/db/s/migration_destination_manager.cpp
@@ -442,27 +442,31 @@ repl::OpTime MigrationDestinationManager::cloneDocumentsFromDonor(
                   "causedBy_exceptionToStatus"_attr = causedBy(redact(exceptionToStatus())));
         }
     }};
-    auto inserterThreadJoinGuard = makeGuard([&] {
-        batches.closeProducerEnd();
-        inserterThread.join();
-    });
 
-    while (true) {
-        opCtx->checkForInterrupt();
 
-        auto res = fetchBatchFn(opCtx);
-
-        opCtx->checkForInterrupt();
-        batches.push(res.getOwned(), opCtx);
-        auto arr = res["objects"].Obj();
-        if (arr.isEmpty()) {
-            inserterThreadJoinGuard.dismiss();
+    {
+        auto inserterThreadJoinGuard = makeGuard([&] {
+            batches.closeProducerEnd();
             inserterThread.join();
-            opCtx->checkForInterrupt();
-            break;
+        });
+
+        while (true) {
+            auto res = fetchBatchFn(opCtx);
+            try {
+                batches.push(res.getOwned(), opCtx);
+                auto arr = res["objects"].Obj();
+                if (arr.isEmpty()) {
+                    break;
+                }
+            } catch (const ExceptionFor<ErrorCodes::ProducerConsumerQueueEndClosed>&) {
+                break;
+            }
         }
-    }
+    }  // This scope ensures that the guard is destroyed
 
+    // This check is necessary because the consumer thread uses killOp to propagate errors to the
+    // producer thread (this thread)
+    opCtx->checkForInterrupt();
     return lastOpApplied;
 }