Moving phase failure deduplication flag to SearchQueryThenFetchAsyncAction

Author: cbuescher

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

@@ -93,7 +93,6 @@ abstract class AbstractSearchAsyncAction<Result extends SearchPhaseResult> exten
     private final Map<String, PendingExecutions> pendingExecutionsPerNode;
     private final AtomicBoolean requestCancelled = new AtomicBoolean();
     private final int skippedCount;
-    private final AtomicBoolean phaseFailureEncountered = new AtomicBoolean();
 
     // protected for tests
     protected final SubscribableListener<Void> doneFuture = new SubscribableListener<>();
@@ -622,10 +621,6 @@ protected BytesReference buildSearchContextId(ShardSearchFailure[] failures) {
      * @param cause the cause of the phase failure
      */
     public void onPhaseFailure(String phase, String msg, Throwable cause) {
-        if (phaseFailureEncountered.compareAndSet(false, true) == false) {
-            // we already encountered a phase failure, so we ignore this one
-            return;
-        }
         raisePhaseFailure(new SearchPhaseExecutionException(phase, msg, cause, buildShardFailures()));
     }
 

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

@@ -462,6 +462,7 @@ protected void doRun(Map<SearchShardIterator, Integer> shardIndexMap) {
                 }
             }
         }
+        final AtomicBoolean phaseFailureEncountered = new AtomicBoolean(false);
         perNodeQueries.forEach((routing, request) -> {
             if (request.shards.size() == 1) {
                 executeAsSingleRequest(routing, request.shards.getFirst());
@@ -483,6 +484,7 @@ protected void doRun(Map<SearchShardIterator, Integer> shardIndexMap) {
             }
             searchTransportService.transportService()
                 .sendChildRequest(connection, NODE_SEARCH_ACTION_NAME, request, task, new TransportResponseHandler<NodeQueryResponse>() {
+
                     @Override
                     public NodeQueryResponse read(StreamInput in) throws IOException {
                         return new NodeQueryResponse(in);
@@ -533,16 +535,21 @@ public void handleException(TransportException e) {
                             if (results instanceof QueryPhaseResultConsumer queryPhaseResultConsumer) {
                                 queryPhaseResultConsumer.failure.compareAndSet(null, cause);
                             }
-                            logger.debug("Raising phase failure for " + cause + " while executing search on node " + routing.nodeId());
-                            onPhaseFailure(getName(), "", cause);
+                            if (phaseFailureEncountered.compareAndSet(false, true)) {
+                                logger.debug("Raising phase failure for " + cause + " while executing search on node " + routing.nodeId());
+                                onPhaseFailure(getName(), "", cause);
+                            } else {
+                                // we already failed the phase, ignore any additional failures and just log them if debug enabled
+                                logger.debug("Ignoring additional phase failure for " + cause + " from search on node " + routing.nodeId());
+                            }
                         }
                     }
                 });
         });
     }
 
     private void executeWithoutBatching(CanMatchPreFilterSearchPhase.SendingTarget targetNode, NodeQueryRequest request) {
-        for (ShardToQuery shard : request.shards) {
+        for (SearchQueryThenFetchAsyncAction.ShardToQuery shard : request.shards) {
             executeAsSingleRequest(targetNode, shard);
         }
     }