Fix assertion error search reduce errors

When receiving exceptions from more than one node on data node search
reduction, SearchQueryThenFetchAsyncAction will raise more than one
phase failure. This can lead to calling the listener in
AbstractSearchAsyncAction more than once, which in turn in tests trips
an assertion in ActionListener#assertFirstRun.

Author: cbuescher

server/src/internalClusterTest/java/org/elasticsearch/search/sort/FieldSortIT.java

@@ -2160,6 +2160,31 @@ public void testLongSortOptimizationCorrectResults() {
         );
     }
 
+    public void testSortMixedFieldTypesSeveralDocs() {
+        assertAcked(
+            prepareCreate("index_long").setMapping("foo", "type=long"),
+            prepareCreate("index_double").setMapping("foo", "type=double")
+        );
+
+        List<IndexRequestBuilder> builders = new ArrayList<>();
+
+        for (int i = 0; i < 10; i++) {
+            builders.add(prepareIndex("index_long").setId(String.valueOf(i)).setSource("foo", i));
+            builders.add(prepareIndex("index_double").setId(String.valueOf(i)).setSource("foo", i));
+        }
+        indexRandom(true, false, builders);
+
+        String errMsg = "Can't sort on field [foo]; the field has incompatible sort types";
+
+        { // mixing long and double types is not allowed
+            SearchPhaseExecutionException exc = expectThrows(
+                SearchPhaseExecutionException.class,
+                prepareSearch("index_long", "index_double").addSort(new FieldSortBuilder("foo")).setSize(20)
+            );
+            assertThat(exc.getCause().toString(), containsString(errMsg));
+        }
+    }
+
     public void testSortMixedFieldTypes() throws IOException {
         assertAcked(
             prepareCreate("index_long").setMapping("foo", "type=long"),

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

@@ -93,6 +93,7 @@ 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<>();
@@ -621,6 +622,10 @@ 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

@@ -533,6 +533,7 @@ 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);
                         }
                     }