SERVER-61633 Join _oplogFetcherExecutor in resharding recipient at exit.

4 files changed, 17 insertions, 0 deletions

diff --git a/src/mongo/db/s/resharding/resharding_data_replication.cpp b/src/mongo/db/s/resharding/resharding_data_replication.cpp
index 8ad0d2a4d3e..3fca4aa7f0a 100644
--- a/src/mongo/db/s/resharding/resharding_data_replication.cpp
+++ b/src/mongo/db/s/resharding/resharding_data_replication.cpp
@@ -411,6 +411,10 @@ void ReshardingDataReplication::shutdown() {
     _oplogFetcherExecutor->shutdown();
 }
 
+void ReshardingDataReplication::join() {
+    _oplogFetcherExecutor->join();
+}
+
 std::vector<NamespaceString> ReshardingDataReplication::ensureStashCollectionsExist(
     OperationContext* opCtx,
     const ChunkManager& sourceChunkMgr,
diff --git a/src/mongo/db/s/resharding/resharding_data_replication.h b/src/mongo/db/s/resharding/resharding_data_replication.h
index fcdd77dcb68..14c31604b0b 100644
--- a/src/mongo/db/s/resharding/resharding_data_replication.h
+++ b/src/mongo/db/s/resharding/resharding_data_replication.h
@@ -126,6 +126,8 @@ public:
     virtual SharedSemiFuture<void> awaitStrictlyConsistent() = 0;
 
     virtual void shutdown() = 0;
+
+    virtual void join() = 0;
 };
 
 class ReshardingDataReplication : public ReshardingDataReplicationInterface {
@@ -169,6 +171,8 @@ public:
 
     void shutdown() override;
 
+    void join() override;
+
     // The following methods are called by ReshardingDataReplication::make() and only exposed
     // publicly for unit-testing purposes.
 
diff --git a/src/mongo/db/s/resharding/resharding_recipient_service.cpp b/src/mongo/db/s/resharding/resharding_recipient_service.cpp
index 380fb5c5616..0e820231a0f 100644
--- a/src/mongo/db/s/resharding/resharding_recipient_service.cpp
+++ b/src/mongo/db/s/resharding/resharding_recipient_service.cpp
@@ -349,6 +349,13 @@ ExecutorFuture<void> ReshardingRecipientService::RecipientStateMachine::_finishR
 
 ExecutorFuture<void> ReshardingRecipientService::RecipientStateMachine::_runMandatoryCleanup(
     Status status, const CancellationToken& stepdownToken) {
+    if (_dataReplication) {
+        // We explicitly shut down and join the ReshardingDataReplication::_oplogFetcherExecutor
+        // because waiting on the _dataReplicationQuiesced future may not do this automatically if
+        // the scoped task executor was already been shut down.
+        _dataReplication->shutdown();
+        _dataReplication->join();
+    }
 
     return _dataReplicationQuiesced.thenRunOn(_recipientService->getInstanceCleanupExecutor())
         .onCompletion([this,
diff --git a/src/mongo/db/s/resharding/resharding_recipient_service_test.cpp b/src/mongo/db/s/resharding/resharding_recipient_service_test.cpp
index f3a0982bed1..d80f9487447 100644
--- a/src/mongo/db/s/resharding/resharding_recipient_service_test.cpp
+++ b/src/mongo/db/s/resharding/resharding_recipient_service_test.cpp
@@ -184,6 +184,8 @@ public:
     };
 
     void shutdown() override {}
+
+    void join() override {}
 };
 
 class ReshardingRecipientServiceForTest : public ReshardingRecipientService {