Changes the API of GatewayAllocator#applyStartedShards and (#20642)

Changes the API of GatewayAllocator#applyStartedShards and 
GatewayAllocator#applyFailedShards to take both a RoutingAllocation
and a list of shards to apply. This allows better mock allocators
to be created as being done in #20637.

Closes #20642

benchmarks/src/main/java/org/elasticsearch/benchmark/routing/allocation/Allocators.java

@@ -24,6 +24,7 @@ import org.elasticsearch.cluster.EmptyClusterInfoService;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.routing.ShardRouting;
 import org.elasticsearch.cluster.routing.allocation.AllocationService;
+import org.elasticsearch.cluster.routing.allocation.FailedShard;
 import org.elasticsearch.cluster.routing.allocation.RoutingAllocation;
 import org.elasticsearch.cluster.routing.allocation.allocator.BalancedShardsAllocator;
 import org.elasticsearch.cluster.routing.allocation.decider.AllocationDecider;
@@ -37,6 +38,7 @@ import org.elasticsearch.gateway.GatewayAllocator;
 import java.lang.reflect.InvocationTargetException;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 
 public final class Allocators {
@@ -48,12 +50,12 @@ public final class Allocators {
         }
 
         @Override
-        public void applyStartedShard(ShardRouting shardRouting) {
+        public void applyStartedShards(RoutingAllocation allocation, List<ShardRouting> startedShards) {
             // noop
         }
 
         @Override
-        public void applyFailedShard(ShardRouting shardRouting) {
+        public void applyFailedShards(RoutingAllocation allocation, List<FailedShard> failedShards) {
             // noop
         }
 

core/src/main/java/org/elasticsearch/cluster/routing/allocation/AllocationService.java

@@ -91,9 +91,7 @@ public class AllocationService extends AbstractComponent {
         RoutingAllocation allocation = new RoutingAllocation(allocationDeciders, routingNodes, clusterState,
             clusterInfoService.getClusterInfo(), currentNanoTime(), false);
         applyStartedShards(allocation, startedShards);
-        for (ShardRouting startedShard : startedShards) {
-            gatewayAllocator.applyStartedShard(startedShard);
-        }
+        gatewayAllocator.applyStartedShards(allocation, startedShards);
         if (withReroute) {
             reroute(allocation);
         }
@@ -172,9 +170,7 @@ public class AllocationService extends AbstractComponent {
                 logger.trace("{} shard routing failed in an earlier iteration (routing: {})", shardToFail.shardId(), shardToFail);
             }
         }
-        for (FailedShard failedShard : failedShards) {
-            gatewayAllocator.applyFailedShard(failedShard.getRoutingEntry());
-        }
+        gatewayAllocator.applyFailedShards(allocation, failedShards);
 
         reroute(allocation);
         String failedShardsAsString = firstListElementsToCommaDelimitedString(failedShards, s -> s.getRoutingEntry().shardId().toString());

core/src/main/java/org/elasticsearch/gateway/GatewayAllocator.java

@@ -28,6 +28,7 @@ import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.routing.RoutingNodes;
 import org.elasticsearch.cluster.routing.RoutingService;
 import org.elasticsearch.cluster.routing.ShardRouting;
+import org.elasticsearch.cluster.routing.allocation.FailedShard;
 import org.elasticsearch.cluster.routing.allocation.RoutingAllocation;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.component.AbstractComponent;
@@ -114,14 +115,18 @@ public class GatewayAllocator extends AbstractComponent {
         return count;
     }
 
-    public void applyStartedShard(final ShardRouting startedShard) {
-        Releasables.close(asyncFetchStarted.remove(startedShard.shardId()));
-        Releasables.close(asyncFetchStore.remove(startedShard.shardId()));
+    public void applyStartedShards(final RoutingAllocation allocation, final List<ShardRouting> startedShards) {
+        for (ShardRouting startedShard : startedShards) {
+            Releasables.close(asyncFetchStarted.remove(startedShard.shardId()));
+            Releasables.close(asyncFetchStore.remove(startedShard.shardId()));
+        }
     }
 
-    public void applyFailedShard(final ShardRouting failedShard) {
-        Releasables.close(asyncFetchStarted.remove(failedShard.shardId()));
-        Releasables.close(asyncFetchStore.remove(failedShard.shardId()));
+    public void applyFailedShards(final RoutingAllocation allocation, final List<FailedShard> failedShards) {
+        for (FailedShard failedShard : failedShards) {
+            Releasables.close(asyncFetchStarted.remove(failedShard.getRoutingEntry().shardId()));
+            Releasables.close(asyncFetchStore.remove(failedShard.getRoutingEntry().shardId()));
+        }
     }
 
     public void allocateUnassigned(final RoutingAllocation allocation) {

test/framework/src/main/java/org/elasticsearch/cluster/ESAllocationTestCase.java

@@ -26,6 +26,7 @@ import org.elasticsearch.cluster.routing.RoutingNodes;
 import org.elasticsearch.cluster.routing.ShardRouting;
 import org.elasticsearch.cluster.routing.UnassignedInfo;
 import org.elasticsearch.cluster.routing.allocation.AllocationService;
+import org.elasticsearch.cluster.routing.allocation.FailedShard;
 import org.elasticsearch.cluster.routing.allocation.RoutingAllocation;
 import org.elasticsearch.cluster.routing.allocation.allocator.BalancedShardsAllocator;
 import org.elasticsearch.cluster.routing.allocation.allocator.ShardsAllocator;
@@ -209,12 +210,12 @@ public abstract class ESAllocationTestCase extends ESTestCase {
         }
 
         @Override
-        public void applyStartedShard(ShardRouting shardRouting) {
+        public void applyStartedShards(RoutingAllocation allocation, List<ShardRouting> startedShards) {
             // no-op
         }
 
         @Override
-        public void applyFailedShard(ShardRouting shardRouting) {
+        public void applyFailedShards(RoutingAllocation allocation, List<FailedShard> failedShards) {
             // no-op
         }
 

test/framework/src/main/java/org/elasticsearch/test/gateway/NoopGatewayAllocator.java

@@ -20,10 +20,13 @@
 package org.elasticsearch.test.gateway;
 
 import org.elasticsearch.cluster.routing.ShardRouting;
+import org.elasticsearch.cluster.routing.allocation.FailedShard;
 import org.elasticsearch.cluster.routing.allocation.RoutingAllocation;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.gateway.GatewayAllocator;
 
+import java.util.List;
+
 /**
  * An allocator used for tests that doesn't do anything
  */
@@ -36,12 +39,12 @@ public class NoopGatewayAllocator extends GatewayAllocator {
     }
 
     @Override
-    public void applyStartedShard(ShardRouting shardRouting) {
+    public void applyStartedShards(RoutingAllocation allocation, List<ShardRouting> startedShards) {
         // noop
     }
 
     @Override
-    public void applyFailedShard(ShardRouting shardRouting) {
+    public void applyFailedShards(RoutingAllocation allocation, List<FailedShard> failedShards) {
         // noop
     }