SERVER-56438 Add ReshardingDonorOplogIterator::dispose() method.

Guarantees that Pipeline underlying ReshardingDonorOplogIterator is
always disposed by ReshardingOplogApplier, even upon cancellation.

1 files changed, 9 insertions, 3 deletions

diff --git a/src/mongo/db/s/resharding/resharding_data_replication.cpp b/src/mongo/db/s/resharding/resharding_data_replication.cpp
index 1a39281774e..dcb344b7502 100644
--- a/src/mongo/db/s/resharding/resharding_data_replication.cpp
+++ b/src/mongo/db/s/resharding/resharding_data_replication.cpp
@@ -288,7 +288,8 @@ SemiFuture<void> ReshardingDataReplication::runUntilStrictlyConsistent(
 
     // Calling _runOplogAppliers() won't actually immediately start performing oplog application.
     // Only after the _startOplogApplication promise is fulfilled will oplog application begin.
-    auto oplogApplierFutures = _runOplogAppliers(executor, errorSource.token(), opCtxFactory);
+    auto oplogApplierFutures =
+        _runOplogAppliers(executor, cleanupExecutor, errorSource.token(), opCtxFactory);
 
     // We must additionally wait for fulfillCloningDoneFuture to become ready to ensure their
     // corresponding promises aren't being fulfilled while the .onCompletion() is running.
@@ -380,6 +381,7 @@ std::vector<SharedSemiFuture<void>> ReshardingDataReplication::_runOplogFetchers
 
 std::vector<SharedSemiFuture<void>> ReshardingDataReplication::_runOplogAppliers(
     std::shared_ptr<executor::TaskExecutor> executor,
+    std::shared_ptr<executor::TaskExecutor> cleanupExecutor,
     CancellationToken cancelToken,
     CancelableOperationContextFactory opCtxFactory) {
     std::vector<SharedSemiFuture<void>> oplogApplierFutures;
@@ -391,8 +393,12 @@ std::vector<SharedSemiFuture<void>> ReshardingDataReplication::_runOplogAppliers
         oplogApplierFutures.emplace_back(
             future_util::withCancellation(_startOplogApplication.getFuture(), cancelToken)
                 .thenRunOn(executor)
-                .then([applier = applier.get(), executor, cancelToken, opCtxFactory] {
-                    return applier->run(executor, cancelToken, opCtxFactory);
+                .then([applier = applier.get(),
+                       executor,
+                       cleanupExecutor,
+                       cancelToken,
+                       opCtxFactory] {
+                    return applier->run(executor, cleanupExecutor, cancelToken, opCtxFactory);
                 })
                 .share());
     }