SERVER-60860 Resharding cloner now really uses read preference Nearest

ReshardingCollectionCloner::_targetAggregationRequest() previously
specified its intended read preference of Nearest on the
AggregateCommandRequest it passes to
sharded_agg_helpers::targetShardsAndAddMergeCursors(). However, when
sharded_agg_helpers::dispatchShardPipeline() is eventually called, this
function ignores the read preference in the command in favor of the read
preference set on the pipeline's expression context's operation context.
As a result, _targetAggregationRequest() now additionally sets its read
preference on the pipeline's opCtx.

(cherry picked from commit cb03a2eb88f90d7ec32dc49fc836c075eb1d44b9)

2 files changed, 9 insertions, 5 deletions

diff --git a/src/mongo/db/s/resharding/resharding_collection_cloner.cpp b/src/mongo/db/s/resharding/resharding_collection_cloner.cpp
index 0225c21cc69..6899a4f6342 100644
--- a/src/mongo/db/s/resharding/resharding_collection_cloner.cpp
+++ b/src/mongo/db/s/resharding/resharding_collection_cloner.cpp
@@ -171,7 +171,8 @@ std::unique_ptr<Pipeline, PipelineDeleter> ReshardingCollectionCloner::makePipel
 }
 
 std::unique_ptr<Pipeline, PipelineDeleter> ReshardingCollectionCloner::_targetAggregationRequest(
-    OperationContext* opCtx, const Pipeline& pipeline) {
+    const Pipeline& pipeline) {
+    auto opCtx = pipeline.getContext()->opCtx;
     // We associate the aggregation cursors established on each donor shard with a logical session
     // to prevent them from killing the cursor when it is idle locally. Due to the cursor's merging
     // behavior across all donor shards, it is possible for the cursor to be active on one donor
@@ -193,7 +194,11 @@ std::unique_ptr<Pipeline, PipelineDeleter> ReshardingCollectionCloner::_targetAg
                                 << repl::readConcernLevels::kSnapshotName
                                 << repl::ReadConcernArgs::kAtClusterTimeFieldName
                                 << _atClusterTime));
-    request.setUnwrappedReadPref(ReadPreferenceSetting{ReadPreference::Nearest}.toContainingBSON());
+    // The read preference on the request is merely informational (e.g. for profiler entries) -- the
+    // pipeline's opCtx setting is actually used when sending the request.
+    auto readPref = ReadPreferenceSetting{ReadPreference::Nearest};
+    request.setUnwrappedReadPref(readPref.toContainingBSON());
+    ReadPreferenceSetting::get(opCtx) = readPref;
 
     return shardVersionRetry(opCtx,
                              Grid::get(opCtx)->catalogCache(),
@@ -227,7 +232,7 @@ std::unique_ptr<Pipeline, PipelineDeleter> ReshardingCollectionCloner::_restartP
     ON_BLOCK_EXIT([curOp] { curOp->done(); });
 
     auto pipeline = _targetAggregationRequest(
-        opCtx, *makePipeline(opCtx, MongoProcessInterface::create(opCtx), idToResumeFrom));
+        *makePipeline(opCtx, MongoProcessInterface::create(opCtx), idToResumeFrom));
 
     if (!idToResumeFrom.missing()) {
         // Skip inserting the first document retrieved after resuming because $gte was used in the
diff --git a/src/mongo/db/s/resharding/resharding_collection_cloner.h b/src/mongo/db/s/resharding/resharding_collection_cloner.h
index 25d737712b0..cf1035ba3a7 100644
--- a/src/mongo/db/s/resharding/resharding_collection_cloner.h
+++ b/src/mongo/db/s/resharding/resharding_collection_cloner.h
@@ -107,8 +107,7 @@ public:
     bool doOneBatch(OperationContext* opCtx, Pipeline& pipeline);
 
 private:
-    std::unique_ptr<Pipeline, PipelineDeleter> _targetAggregationRequest(OperationContext* opCtx,
-                                                                         const Pipeline& pipeline);
+    std::unique_ptr<Pipeline, PipelineDeleter> _targetAggregationRequest(const Pipeline& pipeline);
 
     std::unique_ptr<Pipeline, PipelineDeleter> _restartPipeline(OperationContext* opCtx);