Fix spurious failures in AsyncSearchActionTests

AsyncSearchActionTests#testCleanupOnFailure fails sporadically in CI but not locally.
This commit switches the tests into a SuiteScopeTestCase that creates internal states
once on static members in order to make the tests more reproducible.

Relates #49931

Author: jimczi

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

@@ -128,7 +128,15 @@ private static List<SearchShardIterator> sortShards(GroupShardsIterator<SearchSh
     }
 
     private static boolean shouldSortShards(MinAndMax<?>[] minAndMaxes) {
-        return Arrays.stream(minAndMaxes).anyMatch(Objects::nonNull);
+        Class<?> clazz = null;
+        for (MinAndMax<?> minAndMax : minAndMaxes) {
+            if (clazz == null && minAndMax != null) {
+                clazz = minAndMax.getMin().getClass();
+            } else if (minAndMax != null && clazz != minAndMax.getMax().getClass()) {
+                return false;
+            }
+        }
+        return clazz != null;
     }
 
     private static Comparator<Integer> shardComparator(GroupShardsIterator<SearchShardIterator> shardsIts,

server/src/main/java/org/elasticsearch/search/sort/MinAndMax.java

@@ -55,14 +55,14 @@ public void writeTo(StreamOutput out) throws IOException {
     /**
      * Return the minimum value.
      */
-    T getMin() {
+    public T getMin() {
         return minValue;
     }
 
     /**
      * Return the maximum value.
      */
-    T getMax() {
+    public T getMax() {
         return maxValue;
     }
 

server/src/test/java/org/elasticsearch/action/search/CanMatchPreFilterSearchPhaseTests.java

@@ -18,6 +18,7 @@
  */
 package org.elasticsearch.action.search;
 
+import org.apache.lucene.util.BytesRef;
 import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.OriginalIndices;
@@ -54,6 +55,8 @@
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.IntStream;
 
+import static org.hamcrest.Matchers.equalTo;
+
 public class CanMatchPreFilterSearchPhaseTests extends ESTestCase {
 
     public void testFilterShards() throws InterruptedException {
@@ -350,4 +353,74 @@ public void run() {
             }
         }
     }
+
+    public void testSortShardsDisabled() throws InterruptedException {
+        final TransportSearchAction.SearchTimeProvider timeProvider =
+            new TransportSearchAction.SearchTimeProvider(0, System.nanoTime(), System::nanoTime);
+
+        Map<String, Transport.Connection> lookup = new ConcurrentHashMap<>();
+        DiscoveryNode primaryNode = new DiscoveryNode("node_1", buildNewFakeTransportAddress(), Version.CURRENT);
+        DiscoveryNode replicaNode = new DiscoveryNode("node_2", buildNewFakeTransportAddress(), Version.CURRENT);
+        lookup.put("node1", new SearchAsyncActionTests.MockConnection(primaryNode));
+        lookup.put("node2", new SearchAsyncActionTests.MockConnection(replicaNode));
+
+        for (SortOrder order : SortOrder.values()) {
+            int numShards = randomIntBetween(2, 20);
+            List<ShardId> shardIds = new ArrayList<>();
+            Set<ShardId> shardToSkip = new HashSet<>();
+
+            SearchTransportService searchTransportService = new SearchTransportService(null, null) {
+                @Override
+                public void sendCanMatch(Transport.Connection connection, ShardSearchRequest request, SearchTask task,
+                                         ActionListener<SearchService.CanMatchResponse> listener) {
+                    final MinAndMax<?> minMax;
+                    if (request.shardId().id() == numShards-1) {
+                        minMax = new MinAndMax<>(new BytesRef("bar"), new BytesRef("baz"));
+                    } else {
+                        Long min = rarely() ? null : randomLong();
+                        Long max = min == null ? null : randomLongBetween(min, Long.MAX_VALUE);
+                        minMax = min == null ? null : new MinAndMax<>(min, max);
+                    }
+                    boolean canMatch = frequently();
+                    synchronized (shardIds) {
+                        shardIds.add(request.shardId());
+                        if (canMatch == false) {
+                            shardToSkip.add(request.shardId());
+                        }
+                    }
+                    new Thread(() -> listener.onResponse(new SearchService.CanMatchResponse(canMatch, minMax))).start();
+                }
+            };
+
+            AtomicReference<GroupShardsIterator<SearchShardIterator>> result = new AtomicReference<>();
+            CountDownLatch latch = new CountDownLatch(1);
+            GroupShardsIterator<SearchShardIterator> shardsIter = SearchAsyncActionTests.getShardsIter("logs",
+                new OriginalIndices(new String[]{"logs"}, SearchRequest.DEFAULT_INDICES_OPTIONS),
+                numShards, randomBoolean(), primaryNode, replicaNode);
+            final SearchRequest searchRequest = new SearchRequest();
+            searchRequest.source(new SearchSourceBuilder().sort(SortBuilders.fieldSort("timestamp").order(order)));
+            searchRequest.allowPartialSearchResults(true);
+
+            CanMatchPreFilterSearchPhase canMatchPhase = new CanMatchPreFilterSearchPhase(logger,
+                searchTransportService,
+                (clusterAlias, node) -> lookup.get(node),
+                Collections.singletonMap("_na_", new AliasFilter(null, Strings.EMPTY_ARRAY)),
+                Collections.emptyMap(), Collections.emptyMap(), EsExecutors.newDirectExecutorService(),
+                searchRequest, null, shardsIter, timeProvider, ClusterState.EMPTY_STATE, null,
+                (iter) -> new SearchPhase("test") {
+                    @Override
+                    public void run() {
+                        result.set(iter);
+                        latch.countDown();
+                    }
+                }, SearchResponse.Clusters.EMPTY);
+
+            canMatchPhase.start();
+            latch.await();
+            for (SearchShardIterator i : result.get()) {
+                assertEquals(shardToSkip.contains(i.shardId()), i.skip());
+            }
+            assertThat(result.get().size(), equalTo(numShards));
+        }
+    }
 }

x-pack/plugin/async-search/src/test/java/org/elasticsearch/xpack/search/AsyncSearchActionTests.java

@@ -16,9 +16,9 @@
 import org.elasticsearch.search.aggregations.metrics.InternalMax;
 import org.elasticsearch.search.aggregations.metrics.InternalMin;
 import org.elasticsearch.search.builder.SearchSourceBuilder;
+import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.xpack.core.search.action.AsyncSearchResponse;
 import org.elasticsearch.xpack.core.search.action.SubmitAsyncSearchRequest;
-import org.junit.Before;
 
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -36,18 +36,19 @@
 import static org.hamcrest.Matchers.lessThanOrEqualTo;
 
 // TODO: add tests for keepAlive and expiration
+@ESIntegTestCase.SuiteScopeTestCase
 public class AsyncSearchActionTests extends AsyncSearchIntegTestCase {
-    private String indexName;
-    private int numShards;
-    private int numDocs;
+    private static String indexName;
+    private static int numShards;
+    private static int numDocs;
 
-    private int numKeywords;
-    private Map<String, AtomicInteger> keywordFreqs;
-    private float maxMetric = Float.NEGATIVE_INFINITY;
-    private float minMetric = Float.POSITIVE_INFINITY;
+    private static int numKeywords;
+    private static Map<String, AtomicInteger> keywordFreqs;
+    private static float maxMetric = Float.NEGATIVE_INFINITY;
+    private static float minMetric = Float.POSITIVE_INFINITY;
 
-    @Before
-    public void indexDocuments() throws InterruptedException {
+    @Override
+    public void setupSuiteScopeCluster() throws Exception {
         indexName = "test-async";
         numShards = randomIntBetween(internalCluster().numDataNodes(), internalCluster().numDataNodes()*10);
         numDocs = randomIntBetween(numShards, numShards*10);