Cleanup needlessly holding Exceptions in the can_match phase

Exceptions can be quite expensive heap-wise and we don't use them here any longer,
not even for logging -> replace atomic array with a set of shard indices and
dry up the logic a tiny bit as a result as well.

Author: original-brownbear

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

@@ -38,8 +38,9 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executor;
-import java.util.concurrent.atomic.AtomicReferenceArray;
 import java.util.function.BiFunction;
 
 import static org.elasticsearch.core.Strings.format;
@@ -277,12 +278,11 @@ private Map<SendingTarget, List<SearchShardIterator>> groupByNode(List<SearchSha
     private class Round extends AbstractRunnable {
         private final List<SearchShardIterator> shards;
         private final CountDown countDown;
-        private final AtomicReferenceArray<Exception> failedResponses;
+        private final Set<Integer> failedResponses = ConcurrentHashMap.newKeySet();
 
         Round(List<SearchShardIterator> shards) {
             this.shards = shards;
             this.countDown = new CountDown(shards.size());
-            this.failedResponses = new AtomicReferenceArray<>(shardsIts.size());
         }
 
         @Override
@@ -296,9 +296,7 @@ protected void doRun() {
 
                 if (entry.getKey().nodeId == null) {
                     // no target node: just mark the requests as failed
-                    for (CanMatchNodeRequest.Shard shard : shardLevelRequests) {
-                        onOperationFailed(shard.getShardRequestIndex(), null);
-                    }
+                    onAllFailed(shardLevelRequests);
                     continue;
                 }
 
@@ -321,37 +319,39 @@ public void onResponse(CanMatchNodeResponse canMatchNodeResponse) {
                                     } else {
                                         Exception failure = response.getException();
                                         assert failure != null;
-                                        onOperationFailed(shardLevelRequests.get(i).getShardRequestIndex(), failure);
+                                        onOperationFailed(shardLevelRequests.get(i).getShardRequestIndex());
                                     }
                                 }
                             }
 
                             @Override
                             public void onFailure(Exception e) {
-                                for (CanMatchNodeRequest.Shard shard : shardLevelRequests) {
-                                    onOperationFailed(shard.getShardRequestIndex(), e);
-                                }
+                                onAllFailed(shardLevelRequests);
                             }
                         }
                     );
                 } catch (Exception e) {
-                    for (CanMatchNodeRequest.Shard shard : shardLevelRequests) {
-                        onOperationFailed(shard.getShardRequestIndex(), e);
-                    }
+                    onAllFailed(shardLevelRequests);
                 }
             }
         }
 
+        private void onAllFailed(List<CanMatchNodeRequest.Shard> shardLevelRequests) {
+            for (CanMatchNodeRequest.Shard shard : shardLevelRequests) {
+                onOperationFailed(shard.getShardRequestIndex());
+            }
+        }
+
         private void onOperation(int idx, CanMatchShardResponse response) {
-            failedResponses.set(idx, null);
+            failedResponses.add(idx);
             consumeResult(response);
             if (countDown.countDown()) {
                 finishRound();
             }
         }
 
-        private void onOperationFailed(int idx, Exception e) {
-            failedResponses.set(idx, e);
+        private void onOperationFailed(int idx) {
+            failedResponses.add(idx);
             // we have to carry over shard failures in order to account for them in the response.
             consumeResult(idx, true, null);
             if (countDown.countDown()) {
@@ -363,8 +363,7 @@ private void finishRound() {
             List<SearchShardIterator> remainingShards = new ArrayList<>();
             for (SearchShardIterator ssi : shards) {
                 int shardIndex = shardItIndexMap.get(ssi);
-                Exception failedResponse = failedResponses.get(shardIndex);
-                if (failedResponse != null) {
+                if (failedResponses.contains(shardIndex)) {
                     remainingShards.add(ssi);
                 }
             }