Merge branch 'main' into cuvs-snapshot-update

Author: mayya-sharipova

muted-tests.yml

@@ -726,6 +726,9 @@ tests:
 - class: org.elasticsearch.xpack.esql.ccq.MultiClusterSpecIT
   method: test {csv-spec:fork.ForkWithFiltersOnConstantValues}
   issue: https://github.com/elastic/elasticsearch/issues/136031
+- class: org.elasticsearch.xpack.security.authz.microsoft.MicrosoftGraphAuthzPluginIT
+  method: testConcurrentAuthentication
+  issue: https://github.com/elastic/elasticsearch/issues/135777
 
 # Examples:
 #

server/src/internalClusterTest/java/org/elasticsearch/cluster/routing/allocation/decider/WriteLoadConstraintDeciderIT.java

@@ -28,6 +28,7 @@
 import org.elasticsearch.cluster.routing.ShardRouting;
 import org.elasticsearch.cluster.routing.UnassignedInfo;
 import org.elasticsearch.cluster.routing.allocation.WriteLoadConstraintSettings;
+import org.elasticsearch.cluster.routing.allocation.allocator.BalancedShardsAllocator;
 import org.elasticsearch.cluster.routing.allocation.allocator.DesiredBalanceMetrics;
 import org.elasticsearch.cluster.routing.allocation.allocator.DesiredBalanceShardsAllocator;
 import org.elasticsearch.cluster.service.ClusterService;
@@ -54,14 +55,18 @@
 import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.CountDownLatch;
+import java.util.stream.StreamSupport;
 
 import static java.util.stream.IntStream.range;
 import static org.elasticsearch.cluster.metadata.IndexMetadata.SETTING_NUMBER_OF_REPLICAS;
 import static org.elasticsearch.cluster.metadata.IndexMetadata.SETTING_NUMBER_OF_SHARDS;
+import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.everyItem;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 import static org.hamcrest.Matchers.hasSize;
@@ -130,7 +135,7 @@ public void testHighNodeWriteLoadPreventsNewShardAllocation() {
         setUpMockTransportIndicesStatsResponse(
             harness.firstDiscoveryNode,
             indexMetadata.getNumberOfShards(),
-            createShardStatsResponseForIndex(indexMetadata, harness.randomShardWriteLoad, harness.firstDataNodeId)
+            createShardStatsResponseForIndex(indexMetadata, harness.maxShardWriteLoad, harness.firstDataNodeId)
         );
         setUpMockTransportIndicesStatsResponse(harness.secondDiscoveryNode, 0, List.of());
         setUpMockTransportIndicesStatsResponse(harness.thirdDiscoveryNode, 0, List.of());
@@ -235,7 +240,7 @@ public void testShardsAreAssignedToNotPreferredWhenAlternativeIsNo() {
         setUpMockTransportIndicesStatsResponse(
             harness.firstDiscoveryNode,
             indexMetadata.getNumberOfShards(),
-            createShardStatsResponseForIndex(indexMetadata, harness.randomShardWriteLoad, harness.firstDataNodeId)
+            createShardStatsResponseForIndex(indexMetadata, harness.maxShardWriteLoad, harness.firstDataNodeId)
         );
         setUpMockTransportIndicesStatsResponse(harness.secondDiscoveryNode, 0, List.of());
         setUpMockTransportIndicesStatsResponse(harness.thirdDiscoveryNode, 0, List.of());
@@ -333,7 +338,7 @@ public void testCanRemainNotPreferredIsIgnoredWhenAllOtherNodesReturnNotPreferre
         setUpMockTransportIndicesStatsResponse(
             harness.firstDiscoveryNode,
             indexMetadata.getNumberOfShards(),
-            createShardStatsResponseForIndex(indexMetadata, harness.randomShardWriteLoad, harness.firstDataNodeId)
+            createShardStatsResponseForIndex(indexMetadata, harness.maxShardWriteLoad, harness.firstDataNodeId)
         );
         setUpMockTransportIndicesStatsResponse(harness.secondDiscoveryNode, 0, List.of());
         setUpMockTransportIndicesStatsResponse(harness.thirdDiscoveryNode, 0, List.of());
@@ -429,15 +434,12 @@ public void testCanRemainRelocatesOneShardWhenAHotSpotOccurs() {
          * will show that all shards have non-empty write load stats (so that the WriteLoadDecider will evaluate assigning them to a node).
          */
 
-        IndexMetadata indexMetadata = internalCluster().getCurrentMasterNodeInstance(ClusterService.class)
-            .state()
-            .getMetadata()
-            .getProject()
-            .index(harness.indexName);
+        final ClusterState originalClusterState = internalCluster().getCurrentMasterNodeInstance(ClusterService.class).state();
+        final IndexMetadata indexMetadata = originalClusterState.getMetadata().getProject().index(harness.indexName);
         setUpMockTransportIndicesStatsResponse(
             harness.firstDiscoveryNode,
             indexMetadata.getNumberOfShards(),
-            createShardStatsResponseForIndex(indexMetadata, harness.randomShardWriteLoad, harness.firstDataNodeId)
+            createShardStatsResponseForIndex(indexMetadata, harness.maxShardWriteLoad, harness.firstDataNodeId)
         );
         setUpMockTransportIndicesStatsResponse(harness.secondDiscoveryNode, 0, List.of());
         setUpMockTransportIndicesStatsResponse(harness.thirdDiscoveryNode, 0, List.of());
@@ -483,6 +485,7 @@ public void testCanRemainRelocatesOneShardWhenAHotSpotOccurs() {
                 harness.randomNumberOfShards,
                 countShardsStillAssignedToFirstNode + 1
             );
+            assertThatTheBestShardWasMoved(harness, originalClusterState, desiredBalanceResponse);
         } catch (AssertionError error) {
             ClusterState state = client().admin()
                 .cluster()
@@ -498,6 +501,36 @@ public void testCanRemainRelocatesOneShardWhenAHotSpotOccurs() {
         }
     }
 
+    /**
+     * Determine which shard was moved and check that it's the "best" according to
+     * {@link org.elasticsearch.cluster.routing.allocation.allocator.BalancedShardsAllocator.Balancer.PrioritiseByShardWriteLoadComparator}
+     */
+    private void assertThatTheBestShardWasMoved(
+        TestHarness harness,
+        ClusterState originalClusterState,
+        DesiredBalanceResponse desiredBalanceResponse
+    ) {
+        int movedShardId = desiredBalanceResponse.getRoutingTable().get(harness.indexName).entrySet().stream().filter(e -> {
+            Set<String> desiredNodeIds = e.getValue().desired().nodeIds();
+            return desiredNodeIds.contains(harness.secondDiscoveryNode.getId())
+                || desiredNodeIds.contains(harness.thirdDiscoveryNode.getId());
+        }).findFirst().map(Map.Entry::getKey).orElseThrow(() -> new AssertionError("No shard was moved to a non-hot-spotting node"));
+
+        final BalancedShardsAllocator.Balancer.PrioritiseByShardWriteLoadComparator comparator =
+            new BalancedShardsAllocator.Balancer.PrioritiseByShardWriteLoadComparator(
+                desiredBalanceResponse.getClusterInfo(),
+                originalClusterState.getRoutingNodes().node(harness.firstDataNodeId)
+            );
+
+        final List<ShardRouting> bestShardsToMove = StreamSupport.stream(
+            originalClusterState.getRoutingNodes().node(harness.firstDataNodeId).spliterator(),
+            false
+        ).sorted(comparator).toList();
+
+        // The moved shard should be at the head of the sorted list
+        assertThat(movedShardId, equalTo(bestShardsToMove.get(0).shardId().id()));
+    }
+
     public void testMaxQueueLatencyMetricIsPublished() {
         final Settings settings = Settings.builder()
             .put(
@@ -659,16 +692,35 @@ private NodeUsageStatsForThreadPools createNodeUsageStatsForThreadPools(
     }
 
     /**
-     * Helper to create a list of dummy {@link ShardStats} for the given index, each shard reporting a {@code peakShardWriteLoad} stat.
+     * Helper to create a list of dummy {@link ShardStats} for the given index, each shard being randomly allocated a peak write load
+     * between 0 and {@code maximumShardWriteLoad}. There will always be at least one shard reporting the specified
+     * {@code maximumShardWriteLoad}.
      */
     private List<ShardStats> createShardStatsResponseForIndex(
         IndexMetadata indexMetadata,
-        float peakShardWriteLoad,
+        float maximumShardWriteLoad,
         String assignedShardNodeId
     ) {
-        List<ShardStats> shardStats = new ArrayList<>(indexMetadata.getNumberOfShards());
+        // Randomly distribute shards' peak write-loads so that we can check later that shard movements are prioritized correctly
+        final double writeLoadThreshold = maximumShardWriteLoad
+            * BalancedShardsAllocator.Balancer.PrioritiseByShardWriteLoadComparator.THRESHOLD_RATIO;
+        final List<Double> shardPeakWriteLoads = new ArrayList<>();
+        // Need at least one with the maximum write-load
+        shardPeakWriteLoads.add((double) maximumShardWriteLoad);
+        final int remainingShards = indexMetadata.getNumberOfShards() - 1;
+        // Some over-threshold, some under
+        for (int i = 0; i < remainingShards; ++i) {
+            if (randomBoolean()) {
+                shardPeakWriteLoads.add(randomDoubleBetween(writeLoadThreshold, maximumShardWriteLoad, true));
+            } else {
+                shardPeakWriteLoads.add(randomDoubleBetween(0.0, writeLoadThreshold, true));
+            }
+        }
+        assertThat(shardPeakWriteLoads, hasSize(indexMetadata.getNumberOfShards()));
+        Collections.shuffle(shardPeakWriteLoads, random());
+        final List<ShardStats> shardStats = new ArrayList<>(indexMetadata.getNumberOfShards());
         for (int i = 0; i < indexMetadata.getNumberOfShards(); i++) {
-            shardStats.add(createShardStats(indexMetadata, i, peakShardWriteLoad, assignedShardNodeId));
+            shardStats.add(createShardStats(indexMetadata, i, shardPeakWriteLoads.get(i), assignedShardNodeId));
         }
         return shardStats;
     }
@@ -719,7 +771,7 @@ private TestHarness setUpThreeTestNodesAndAllIndexShardsOnFirstNode() {
         int randomUtilizationThresholdPercent = randomIntBetween(50, 100);
         int randomNumberOfWritePoolThreads = randomIntBetween(2, 20);
         long randomQueueLatencyThresholdMillis = randomLongBetween(1, 20_000);
-        float randomShardWriteLoad = randomFloatBetween(0.0f, 0.01f, false);
+        float maximumShardWriteLoad = randomFloatBetween(0.0f, 0.01f, false);
         Settings settings = enabledWriteLoadDeciderSettings(randomUtilizationThresholdPercent, randomQueueLatencyThresholdMillis);
 
         internalCluster().startMasterOnlyNode(settings);
@@ -756,8 +808,8 @@ private TestHarness setUpThreeTestNodesAndAllIndexShardsOnFirstNode() {
                 + randomUtilizationThresholdPercent
                 + ",  write threads: "
                 + randomNumberOfWritePoolThreads
-                + ", individual shard write loads: "
-                + randomShardWriteLoad
+                + ", maximum shard write load: "
+                + maximumShardWriteLoad
         );
 
         /**
@@ -775,7 +827,7 @@ private TestHarness setUpThreeTestNodesAndAllIndexShardsOnFirstNode() {
 
         // Calculate the maximum utilization a node can report while still being able to accept all relocating shards
         int shardWriteLoadOverhead = shardLoadUtilizationOverhead(
-            randomShardWriteLoad * randomNumberOfShards,
+            maximumShardWriteLoad * randomNumberOfShards,
             randomNumberOfWritePoolThreads
         );
         int maxUtilBelowThresholdThatAllowsAllShardsToRelocate = randomUtilizationThresholdPercent - shardWriteLoadOverhead - 1;
@@ -819,7 +871,7 @@ private TestHarness setUpThreeTestNodesAndAllIndexShardsOnFirstNode() {
             randomUtilizationThresholdPercent,
             randomNumberOfWritePoolThreads,
             randomQueueLatencyThresholdMillis,
-            randomShardWriteLoad,
+            maximumShardWriteLoad,
             indexName,
             randomNumberOfShards,
             maxUtilBelowThresholdThatAllowsAllShardsToRelocate
@@ -842,7 +894,7 @@ record TestHarness(
         int randomUtilizationThresholdPercent,
         int randomNumberOfWritePoolThreads,
         long randomQueueLatencyThresholdMillis,
-        float randomShardWriteLoad,
+        float maxShardWriteLoad,
         String indexName,
         int randomNumberOfShards,
         int maxUtilBelowThresholdThatAllowsAllShardsToRelocate

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -53,8 +53,6 @@ static TransportVersion def(int id) {
     }
 
     // TODO: ES-10337 we can remove all transport versions earlier than 8.18
-    public static final TransportVersion V_7_1_0 = def(7_01_00_99);
-    public static final TransportVersion V_7_2_0 = def(7_02_00_99);
     public static final TransportVersion V_7_3_0 = def(7_03_00_99);
     public static final TransportVersion V_7_3_2 = def(7_03_02_99);
     public static final TransportVersion V_7_4_0 = def(7_04_00_99);

server/src/main/java/org/elasticsearch/action/support/replication/ReplicationRequest.java

@@ -9,6 +9,7 @@
 
 package org.elasticsearch.action.support.replication;
 
+import org.elasticsearch.TransportVersion;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.IndicesRequest;
 import org.elasticsearch.action.LegacyActionRequest;
@@ -40,6 +41,11 @@ public abstract class ReplicationRequest<Request extends ReplicationRequest<Requ
 
     public static final TimeValue DEFAULT_TIMEOUT = TimeValue.timeValueMinutes(1);
 
+    // superseded
+    private static final TransportVersion INDEX_RESHARD_SHARDCOUNT_SUMMARY = TransportVersion.fromName("index_reshard_shardcount_summary");
+    // bumped to use VInt instead of Int
+    private static final TransportVersion INDEX_RESHARD_SHARDCOUNT_SMALL = TransportVersion.fromName("index_reshard_shardcount_small");
+
     /**
      * Target shard the request should execute on. In case of index and delete requests,
      * shard id gets resolved by the transport action before performing request operation
@@ -51,41 +57,8 @@ public abstract class ReplicationRequest<Request extends ReplicationRequest<Requ
     protected String index;
 
     /**
-     * The reshardSplitShardCountSummary has been added to accommodate the Resharding feature.
-     * This is populated when the coordinator is deciding which shards a request applies to.
-     * For example, {@link org.elasticsearch.action.bulk.BulkOperation} splits
-     * an incoming bulk request into shard level {@link org.elasticsearch.action.bulk.BulkShardRequest}
-     * based on its cluster state view of the number of shards that are ready for indexing.
-     * The purpose of this metadata is to reconcile the cluster state visible at the coordinating
-     * node with that visible at the source shard node. (w.r.t resharding).
-     * When an index is being split, there is a point in time when the newly created shard (target shard)
-     * takes over its portion of the document space from the original shard (source shard).
-     * Although the handoff is atomic at the original (source shard) and new shards (target shard),
-     * there is a window of time between the coordinating node creating a shard request and the shard receiving and processing it.
-     * This field is used by the original shard (source shard) when it processes the request to detect whether
-     * the coordinator's view of the new shard's state when it created the request matches the shard's current state,
-     * or whether the request must be reprocessed taking into account the current shard states.
-     *
-     * Note that we are able to get away with a single number, instead of an array of target shard states,
-     * because we only allow splits in increments of 2x.
-     *
-     * Example 1:
-     * Suppose we are resharding an index from 2 -> 4 shards. While splitting a bulk request, the coordinator observes
-     * that target shards are not ready for indexing. So requests that are meant for shard 0 and 2 are bundled together,
-     * sent to shard 0 with “reshardSplitShardCountSummary” 2 in the request.
-     * Requests that are meant for shard 1 and 3 are bundled together,
-     * sent to shard 1 with “reshardSplitShardCountSummary” 2 in the request.
-     *
-     * Example 2:
-     * Suppose we are resharding an index from 4 -> 8 shards. While splitting a bulk request, the coordinator observes
-     * that source shard 0 has completed HANDOFF but source shards 1, 2, 3 have not completed handoff.
-     * So, the shard-bulk-request it sends to shard 0 and 4 has the "reshardSplitShardCountSummary" 8,
-     * while the shard-bulk-request it sends to shard 1,2,3 has the "reshardSplitShardCountSummary" 4.
-     * Note that in this case no shard-bulk-request is sent to shards 5, 6, 7 and the requests that were meant for these target shards
-     * are bundled together with and sent to their source shards.
-     *
-     * A value of 0 indicates an INVALID reshardSplitShardCountSummary. Hence, a request with INVALID reshardSplitShardCountSummary
-     * will be treated as a Summary mismatch on the source shard node.
+     * The reshardSplitShardCountSummary has been added to support in-place resharding.
+     * See {@link SplitShardCountSummary} for details.
      */
     protected final SplitShardCountSummary reshardSplitShardCountSummary;
 
@@ -128,7 +101,13 @@ public ReplicationRequest(@Nullable ShardId shardId, SplitShardCountSummary resh
         if (thinRead) {
             this.reshardSplitShardCountSummary = reshardSplitShardCountSummary;
         } else {
-            this.reshardSplitShardCountSummary = new SplitShardCountSummary(in);
+            if (in.getTransportVersion().supports(INDEX_RESHARD_SHARDCOUNT_SMALL)) {
+                this.reshardSplitShardCountSummary = SplitShardCountSummary.fromInt(in.readVInt());
+            } else if (in.getTransportVersion().supports(INDEX_RESHARD_SHARDCOUNT_SUMMARY)) {
+                this.reshardSplitShardCountSummary = SplitShardCountSummary.fromInt(in.readInt());
+            } else {
+                this.reshardSplitShardCountSummary = SplitShardCountSummary.UNSET;
+            }
         }
     }
 
@@ -257,7 +236,11 @@ public void writeTo(StreamOutput out) throws IOException {
         out.writeTimeValue(timeout);
         out.writeString(index);
         out.writeVLong(routedBasedOnClusterVersion);
-        reshardSplitShardCountSummary.writeTo(out);
+        if (out.getTransportVersion().supports(INDEX_RESHARD_SHARDCOUNT_SMALL)) {
+            out.writeVInt(reshardSplitShardCountSummary.asInt());
+        } else if (out.getTransportVersion().supports(INDEX_RESHARD_SHARDCOUNT_SUMMARY)) {
+            out.writeInt(reshardSplitShardCountSummary.asInt());
+        }
     }
 
     /**