Clamp auto-expand replicas to the closest value

Ensure that the number of replicas chosen for an auto-expand-able shard
is within the range of the available data nodes, i.e., excluding those
nodes that cannot be assigned a replica.

Closes elastic#84788

Author: pxsalehi

server/src/main/java/org/elasticsearch/cluster/metadata/AutoExpandReplicas.java

@@ -18,7 +18,6 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.OptionalInt;
 import java.util.function.Supplier;
 
 import static org.elasticsearch.cluster.metadata.MetadataIndexStateService.isIndexVerifiedBeforeClosed;
@@ -101,7 +100,7 @@ public boolean expandToAllNodes() {
         return maxReplicas == Integer.MAX_VALUE;
     }
 
-    public OptionalInt getDesiredNumberOfReplicas(IndexMetadata indexMetadata, RoutingAllocation allocation) {
+    public int getDesiredNumberOfReplicas(IndexMetadata indexMetadata, RoutingAllocation allocation) {
         assert enabled : "should only be called when enabled";
         int numMatchingDataNodes = 0;
         for (DiscoveryNode discoveryNode : allocation.nodes().getDataNodes().values()) {
@@ -110,20 +109,21 @@ public OptionalInt getDesiredNumberOfReplicas(IndexMetadata indexMetadata, Routi
                 numMatchingDataNodes++;
             }
         }
+        return calculateDesiredNumberOfReplicas(numMatchingDataNodes);
+    }
 
+    // Package private only for testing purposes!
+    int calculateDesiredNumberOfReplicas(int numMatchingDataNodes) {
         final int min = minReplicas();
         final int max = getMaxReplicas(numMatchingDataNodes);
         int numberOfReplicas = numMatchingDataNodes - 1;
+        // Make sure number of replicas is always between min and max
         if (numberOfReplicas < min) {
             numberOfReplicas = min;
         } else if (numberOfReplicas > max) {
             numberOfReplicas = max;
         }
-
-        if (numberOfReplicas >= min && numberOfReplicas <= max) {
-            return OptionalInt.of(numberOfReplicas);
-        }
-        return OptionalInt.empty();
+        return numberOfReplicas;
     }
 
     @Override
@@ -153,11 +153,10 @@ public static Map<Integer, List<String>> getAutoExpandReplicaChanges(
                 if (allocation == null) {
                     allocation = allocationSupplier.get();
                 }
-                autoExpandReplicas.getDesiredNumberOfReplicas(indexMetadata, allocation).ifPresent(numberOfReplicas -> {
-                    if (numberOfReplicas != indexMetadata.getNumberOfReplicas()) {
-                        nrReplicasChanged.computeIfAbsent(numberOfReplicas, ArrayList::new).add(indexMetadata.getIndex().getName());
-                    }
-                });
+                int numberOfReplicas = autoExpandReplicas.getDesiredNumberOfReplicas(indexMetadata, allocation);
+                if (numberOfReplicas != indexMetadata.getNumberOfReplicas()) {
+                    nrReplicasChanged.computeIfAbsent(numberOfReplicas, ArrayList::new).add(indexMetadata.getIndex().getName());
+                }
             }
         }
         return nrReplicasChanged;

server/src/test/java/org/elasticsearch/cluster/metadata/AutoExpandReplicasTests.java

@@ -282,4 +282,16 @@ public void testOnlyAutoExpandAllocationFilteringAfterAllNodesUpgraded() {
             terminate(threadPool);
         }
     }
+
+    public void testCalculateDesiredNumberOfReplicas() {
+        String settingValue = randomFrom("0-all", "0-3");
+        AutoExpandReplicas autoExpandReplicas = AutoExpandReplicas.SETTING.get(
+            Settings.builder().put(SETTING_AUTO_EXPAND_REPLICAS, settingValue).build()
+        );
+        int max = autoExpandReplicas.maxReplicas();
+        assertThat(autoExpandReplicas.calculateDesiredNumberOfReplicas(0), equalTo(0));
+        assertThat(autoExpandReplicas.calculateDesiredNumberOfReplicas(1), equalTo(0));
+        assertThat(autoExpandReplicas.calculateDesiredNumberOfReplicas(2), equalTo(1));
+        assertThat(autoExpandReplicas.calculateDesiredNumberOfReplicas(max), equalTo(max - 1));
+    }
 }