Reduce latency in fetch phase for large shard counts (elastic#120419)

We can reduce the latency of the fetch response by a non-trivial amount by moving
the context-freeing for irrelevant shards to the end of the forked action.

For large shard counts (and/or with security in the mix) the old comment is not entirely
correct and waking up the selector many times over + doing some authz work might consume
macroscopic time.
Moving the freeing to the end of the task causes the fetches to be sent out potentially much quicker,
reduces contention on the counter and reduces the impact of potential head-of-line blocking issues
on the connections that might see context freeing needlessly queue after actual fetches.

Author: original-brownbear

server/src/main/java/org/elasticsearch/action/search/FetchSearchPhase.java

@@ -158,30 +158,31 @@ private void innerRunFetch(ScoreDoc[] scoreDocs, int numShards, SearchPhaseContr
         );
         for (int i = 0; i < docIdsToLoad.length; i++) {
             List<Integer> entry = docIdsToLoad[i];
-            RankDocShardInfo rankDocs = rankDocsPerShard == null || rankDocsPerShard.get(i).isEmpty()
-                ? null
-                : new RankDocShardInfo(rankDocsPerShard.get(i));
-            SearchPhaseResult shardPhaseResult = searchPhaseShardResults.get(i);
             if (entry == null) { // no results for this shard ID
-                if (shardPhaseResult != null) {
-                    // if we got some hits from this shard we have to release the context there
-                    // we do this as we go since it will free up resources and passing on the request on the
-                    // transport layer is cheap.
-                    releaseIrrelevantSearchContext(shardPhaseResult, context);
-                    progressListener.notifyFetchResult(i);
-                }
+                // if we got some hits from this shard we have to release the context
+                // we do this below after sending out the fetch requests relevant to the search to give priority to those requests
+                // that contribute to the final search response
                 // in any case we count down this result since we don't talk to this shard anymore
                 counter.countDown();
             } else {
                 executeFetch(
-                    shardPhaseResult,
+                    searchPhaseShardResults.get(i),
                     counter,
                     entry,
-                    rankDocs,
+                    rankDocsPerShard == null || rankDocsPerShard.get(i).isEmpty() ? null : new RankDocShardInfo(rankDocsPerShard.get(i)),
                     (lastEmittedDocPerShard != null) ? lastEmittedDocPerShard[i] : null
                 );
             }
         }
+        for (int i = 0; i < docIdsToLoad.length; i++) {
+            if (docIdsToLoad[i] == null) {
+                SearchPhaseResult shardPhaseResult = searchPhaseShardResults.get(i);
+                if (shardPhaseResult != null) {
+                    releaseIrrelevantSearchContext(shardPhaseResult, context);
+                    progressListener.notifyFetchResult(i);
+                }
+            }
+        }
     }
 
     private List<Map<Integer, RankDoc>> splitRankDocsPerShard(ScoreDoc[] scoreDocs, int numShards) {