Merge branch 'main' into feature/unmapped_fields_squashed

Author: GalLalouche

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

@@ -90,15 +90,13 @@ abstract class AbstractSearchAsyncAction<Result extends SearchPhaseResult> exten
     private final Object shardFailuresMutex = new Object();
     private final AtomicBoolean hasShardResponse = new AtomicBoolean(false);
     private final AtomicInteger successfulOps = new AtomicInteger();
-    private final AtomicInteger skippedOps = new AtomicInteger();
     private final SearchTimeProvider timeProvider;
     private final SearchResponse.Clusters clusters;
 
     protected final GroupShardsIterator<SearchShardIterator> toSkipShardsIts;
     protected final GroupShardsIterator<SearchShardIterator> shardsIts;
     private final SearchShardIterator[] shardIterators;
-    private final int expectedTotalOps;
-    private final AtomicInteger totalOps = new AtomicInteger();
+    private final AtomicInteger outstandingShards;
     private final int maxConcurrentRequestsPerNode;
     private final Map<String, PendingExecutions> pendingExecutionsPerNode = new ConcurrentHashMap<>();
     private final boolean throttleConcurrentRequests;
@@ -139,18 +137,12 @@ abstract class AbstractSearchAsyncAction<Result extends SearchPhaseResult> exten
         }
         this.toSkipShardsIts = new GroupShardsIterator<>(toSkipIterators);
         this.shardsIts = new GroupShardsIterator<>(iterators);
-
+        outstandingShards = new AtomicInteger(shardsIts.size());
         this.shardIterators = iterators.toArray(new SearchShardIterator[0]);
         // we later compute the shard index based on the natural order of the shards
         // that participate in the search request. This means that this number is
         // consistent between two requests that target the same shards.
         Arrays.sort(shardIterators);
-
-        // we need to add 1 for non active partition, since we count it in the total. This means for each shard in the iterator we sum up
-        // it's number of active shards but use 1 as the default if no replica of a shard is active at this point.
-        // on a per shards level we use shardIt.remaining() to increment the totalOps pointer but add 1 for the current shard result
-        // we process hence we add one for the non active partition here.
-        this.expectedTotalOps = shardsIts.totalSizeWith1ForEmpty();
         this.maxConcurrentRequestsPerNode = maxConcurrentRequestsPerNode;
         // in the case were we have less shards than maxConcurrentRequestsPerNode we don't need to throttle
         this.throttleConcurrentRequests = maxConcurrentRequestsPerNode < shardsIts.size();
@@ -251,9 +243,8 @@ protected final void run() {
 
     void skipShard(SearchShardIterator iterator) {
         successfulOps.incrementAndGet();
-        skippedOps.incrementAndGet();
         assert iterator.skip();
-        successfulShardExecution(iterator);
+        successfulShardExecution();
     }
 
     private static boolean assertExecuteOnStartThread() {
@@ -380,7 +371,7 @@ protected void executeNextPhase(String currentPhase, Supplier<SearchPhase> nextP
                             "Partial shards failure (unavailable: {}, successful: {}, skipped: {}, num-shards: {}, phase: {})",
                             discrepancy,
                             successfulOps.get(),
-                            skippedOps.get(),
+                            toSkipShardsIts.size(),
                             getNumShards(),
                             currentPhase
                         );
@@ -449,17 +440,14 @@ private void onShardFailure(final int shardIndex, SearchShardTarget shard, final
             }
             onShardGroupFailure(shardIndex, shard, e);
         }
-        final int totalOps = this.totalOps.incrementAndGet();
-        if (totalOps == expectedTotalOps) {
-            onPhaseDone();
-        } else if (totalOps > expectedTotalOps) {
-            throw new AssertionError(
-                "unexpected higher total ops [" + totalOps + "] compared to expected [" + expectedTotalOps + "]",
-                new SearchPhaseExecutionException(getName(), "Shard failures", null, buildShardFailures())
-            );
+        if (lastShard == false) {
+            performPhaseOnShard(shardIndex, shardIt, nextShard);
         } else {
-            if (lastShard == false) {
-                performPhaseOnShard(shardIndex, shardIt, nextShard);
+            // count down outstanding shards, we're done with this shard as there's no more copies to try
+            final int outstanding = outstandingShards.decrementAndGet();
+            assert outstanding >= 0 : "outstanding: " + outstanding;
+            if (outstanding == 0) {
+                onPhaseDone();
             }
         }
     }
@@ -535,10 +523,10 @@ protected void onShardResult(Result result, SearchShardIterator shardIt) {
         if (logger.isTraceEnabled()) {
             logger.trace("got first-phase result from {}", result != null ? result.getSearchShardTarget() : null);
         }
-        results.consumeResult(result, () -> onShardResultConsumed(result, shardIt));
+        results.consumeResult(result, () -> onShardResultConsumed(result));
     }
 
-    private void onShardResultConsumed(Result result, SearchShardIterator shardIt) {
+    private void onShardResultConsumed(Result result) {
         successfulOps.incrementAndGet();
         // clean a previous error on this shard group (note, this code will be serialized on the same shardIndex value level
         // so its ok concurrency wise to miss potentially the shard failures being created because of another failure
@@ -552,28 +540,14 @@ private void onShardResultConsumed(Result result, SearchShardIterator shardIt) {
         // cause the successor to read a wrong value from successfulOps if second phase is very fast ie. count etc.
         // increment all the "future" shards to update the total ops since we some may work and some may not...
         // and when that happens, we break on total ops, so we must maintain them
-        successfulShardExecution(shardIt);
+        successfulShardExecution();
     }
 
-    private void successfulShardExecution(SearchShardIterator shardsIt) {
-        final int remainingOpsOnIterator;
-        if (shardsIt.skip()) {
-            // It's possible that we're skipping a shard that's unavailable
-            // but its range was available in the IndexMetadata, in that
-            // case the shardsIt.remaining() would be 0, expectedTotalOps
-            // accounts for unavailable shards too.
-            remainingOpsOnIterator = Math.max(shardsIt.remaining(), 1);
-        } else {
-            remainingOpsOnIterator = shardsIt.remaining() + 1;
-        }
-        final int xTotalOps = totalOps.addAndGet(remainingOpsOnIterator);
-        if (xTotalOps == expectedTotalOps) {
+    private void successfulShardExecution() {
+        final int outstanding = outstandingShards.decrementAndGet();
+        assert outstanding >= 0 : "outstanding: " + outstanding;
+        if (outstanding == 0) {
             onPhaseDone();
-        } else if (xTotalOps > expectedTotalOps) {
-            throw new AssertionError(
-                "unexpected higher total ops [" + xTotalOps + "] compared to expected [" + expectedTotalOps + "]",
-                new SearchPhaseExecutionException(getName(), "Shard failures", null, buildShardFailures())
-            );
         }
     }
 
@@ -640,7 +614,7 @@ private SearchResponse buildSearchResponse(
             scrollId,
             getNumShards(),
             numSuccess,
-            skippedOps.get(),
+            toSkipShardsIts.size(),
             buildTookInMillis(),
             failures,
             clusters,

server/src/main/java/org/elasticsearch/cluster/routing/GroupShardsIterator.java

@@ -41,26 +41,6 @@ public GroupShardsIterator(List<ShardIt> iterators) {
         this.iterators = iterators;
     }
 
-    /**
-     * Returns the total number of shards within all groups
-     * @return total number of shards
-     */
-    public int totalSize() {
-        return iterators.stream().mapToInt(Countable::size).sum();
-    }
-
-    /**
-     * Returns the total number of shards plus the number of empty groups
-     * @return number of shards and empty groups
-     */
-    public int totalSizeWith1ForEmpty() {
-        int size = 0;
-        for (ShardIt shard : iterators) {
-            size += Math.max(1, shard.size());
-        }
-        return size;
-    }
-
     /**
      * Return the number of groups
      * @return number of groups

server/src/main/java/org/elasticsearch/transport/TransportHandshaker.java

@@ -44,49 +44,17 @@ final class TransportHandshaker {
      * ignores the body of the request. After the handshake, the OutboundHandler uses the min(local,remote) protocol version for all later
      * messages.
      *
-     * This version supports three handshake protocols, v6080099, v7170099 and v8800000, which respectively have the same message structure
-     * as the transport protocols of v6.8.0, v7.17.0, and v8.18.0. This node only sends v7170099 requests, but it can send a valid response
-     * to any v6080099 or v8800000 requests that it receives.
+     * This version supports two handshake protocols, v7170099 and v8800000, which respectively have the same message structure as the
+     * transport protocols of v7.17.0, and v8.18.0. This node only sends v8800000 requests, but it can send a valid response to any v7170099
+     * requests that it receives.
      *
      * Note that these are not really TransportVersion constants as used elsewhere in ES, they're independent things that just happen to be
-     * stored in the same location in the message header and which roughly match the same ID numbering scheme. Older versions of ES did
-     * rely on them matching the real transport protocol (which itself matched the release version numbers), but these days that's no longer
+     * stored in the same location in the message header and which roughly match the same ID numbering scheme. Older versions of ES did rely
+     * on them matching the real transport protocol (which itself matched the release version numbers), but these days that's no longer
      * true.
      *
      * Here are some example messages, broken down to show their structure. See TransportHandshakerRawMessageTests for supporting tests.
      *
-     * ## v6080099 Request:
-     *
-     * 45 53                            -- 'ES' marker
-     * 00 00 00 34                      -- total message length
-     *    00 00 00 00 00 00 00 01       -- request ID
-     *    08                            -- status flags (0b1000 == handshake request)
-     *    00 5c c6 63                   -- handshake protocol version (0x5cc663 == 6080099)
-     *    00                            -- no request headers [1]
-     *    00                            -- no response headers [1]
-     *    01                            -- one feature [2]
-     *       06                         -- feature name length
-     *          78 2d 70 61 63 6b       -- feature name 'x-pack'
-     *    16                            -- action string size
-     *       69 6e 74 65 72 6e 61 6c    }
-     *       3a 74 63 70 2f 68 61 6e    }- ASCII representation of HANDSHAKE_ACTION_NAME
-     *       64 73 68 61 6b 65          }
-     *    00                            -- no parent task ID [3]
-     *    04                            -- payload length
-     *       8b d5 b5 03                -- max acceptable protocol version (vInt: 00000011 10110101 11010101 10001011 == 7170699)
-     *
-     * ## v6080099 Response:
-     *
-     * 45 53                            -- 'ES' marker
-     * 00 00 00 13                      -- total message length
-     *    00 00 00 00 00 00 00 01       -- request ID (copied from request)
-     *    09                            -- status flags (0b1001 == handshake response)
-     *    00 5c c6 63                   -- handshake protocol version (0x5cc663 == 6080099, copied from request)
-     *    00                            -- no request headers [1]
-     *    00                            -- no response headers [1]
-     *    c3 f9 eb 03                   -- max acceptable protocol version (vInt: 00000011 11101011 11111001 11000011 == 8060099)
-     *
-     *
      * ## v7170099 Requests:
      *
      * 45 53                            -- 'ES' marker
@@ -158,14 +126,9 @@ final class TransportHandshaker {
      * [3] Parent task ID should be empty; see org.elasticsearch.tasks.TaskId.writeTo for its structure.
      */
 
-    static final TransportVersion V7_HANDSHAKE_VERSION = TransportVersion.fromId(6_08_00_99);
     static final TransportVersion V8_HANDSHAKE_VERSION = TransportVersion.fromId(7_17_00_99);
     static final TransportVersion V9_HANDSHAKE_VERSION = TransportVersion.fromId(8_800_00_0);
-    static final Set<TransportVersion> ALLOWED_HANDSHAKE_VERSIONS = Set.of(
-        V7_HANDSHAKE_VERSION,
-        V8_HANDSHAKE_VERSION,
-        V9_HANDSHAKE_VERSION
-    );
+    static final Set<TransportVersion> ALLOWED_HANDSHAKE_VERSIONS = Set.of(V8_HANDSHAKE_VERSION, V9_HANDSHAKE_VERSION);
 
     static final String HANDSHAKE_ACTION_NAME = "internal:tcp/handshake";
     private final ConcurrentMap<Long, HandshakeResponseHandler> pendingHandshakes = new ConcurrentHashMap<>();
@@ -203,7 +166,7 @@ void sendHandshake(
         );
         boolean success = false;
         try {
-            handshakeRequestSender.sendRequest(node, channel, requestId, V8_HANDSHAKE_VERSION);
+            handshakeRequestSender.sendRequest(node, channel, requestId, V9_HANDSHAKE_VERSION);
 
             threadPool.schedule(
                 () -> handler.handleLocalException(new ConnectTransportException(node, "handshake_timeout[" + timeout + "]")),

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

@@ -144,7 +144,6 @@ protected void run() {
                     hits.decRef();
                 }
             } finally {
-                mockSearchPhaseContext.execute(() -> {});
                 var resp = mockSearchPhaseContext.searchResponse.get();
                 if (resp != null) {
                     resp.decRef();

server/src/test/java/org/elasticsearch/cluster/routing/GroupShardsIteratorTests.java

@@ -38,33 +38,6 @@ private static List<ShardRouting> randomShardRoutings(ShardId shardId, int numRe
         return shardRoutings;
     }
 
-    public void testSize() {
-        List<ShardIterator> list = new ArrayList<>();
-        Index index = new Index("foo", "na");
-        {
-            ShardId shardId = new ShardId(index, 0);
-            list.add(new PlainShardIterator(shardId, randomShardRoutings(shardId, 2)));
-        }
-        list.add(new PlainShardIterator(new ShardId(index, 1), Collections.emptyList()));
-        {
-            ShardId shardId = new ShardId(index, 2);
-            list.add(new PlainShardIterator(shardId, randomShardRoutings(shardId, 0)));
-        }
-        index = new Index("foo_1", "na");
-        {
-            ShardId shardId = new ShardId(index, 0);
-            list.add(new PlainShardIterator(shardId, randomShardRoutings(shardId, 0)));
-        }
-        {
-            ShardId shardId = new ShardId(index, 1);
-            list.add(new PlainShardIterator(shardId, randomShardRoutings(shardId, 0)));
-        }
-        GroupShardsIterator<ShardIterator> iter = new GroupShardsIterator<>(list);
-        assertEquals(7, iter.totalSizeWith1ForEmpty());
-        assertEquals(5, iter.size());
-        assertEquals(6, iter.totalSize());
-    }
-
     public void testIterate() {
         List<ShardIterator> list = new ArrayList<>();
         Index index = new Index("foo", "na");