Reduce cluster update reroutes with async fetch

kimchy · kimchy · commit 5fa95206fdaf · 2015-05-29T13:53:15.000-07:00
When using async fetch, we can end up with cluster updates and reroutes based on teh number of shards. While not disastrous we can optimize it, since a single reroute is enough to apply to all the async fetch results that arrived during that time.
diff --git a/src/main/java/org/elasticsearch/gateway/local/LocalGatewayAllocator.java b/src/main/java/org/elasticsearch/gateway/local/LocalGatewayAllocator.java
@@ -59,6 +59,7 @@
 
 import java.util.*;
 import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
  *
@@ -172,7 +173,7 @@ public boolean allocateUnassigned(RoutingAllocation allocation) {
 
             AsyncShardFetch<TransportNodesListGatewayStartedShards.NodeLocalGatewayStartedShards> fetch = asyncFetchStarted.get(shard.shardId());
             if (fetch == null) {
-                fetch = new InternalAsyncFetch<>(logger, "shard_started", shard.shardId(), startedAction, clusterService, allocationService);
+                fetch = new InternalAsyncFetch<>(logger, "shard_started", shard.shardId(), startedAction);
                 asyncFetchStarted.put(shard.shardId(), fetch);
             }
             AsyncShardFetch.FetchResult<TransportNodesListGatewayStartedShards.NodeLocalGatewayStartedShards> shardState = fetch.fetchData(nodes, metaData, allocation.getIgnoreNodes(shard.shardId()));
@@ -408,7 +409,7 @@ public int compare(DiscoveryNode o1, DiscoveryNode o2) {
 
             AsyncShardFetch<TransportNodesListShardStoreMetaData.NodeStoreFilesMetaData> fetch = asyncFetchStore.get(shard.shardId());
             if (fetch == null) {
-                fetch = new InternalAsyncFetch<>(logger, "shard_store", shard.shardId(), storeAction, clusterService, allocationService);
+                fetch = new InternalAsyncFetch<>(logger, "shard_store", shard.shardId(), storeAction);
                 asyncFetchStore.put(shard.shardId(), fetch);
             }
             AsyncShardFetch.FetchResult<TransportNodesListShardStoreMetaData.NodeStoreFilesMetaData> shardStores = fetch.fetchData(nodes, metaData, allocation.getIgnoreNodes(shard.shardId()));
@@ -519,23 +520,24 @@ public int compare(DiscoveryNode o1, DiscoveryNode o2) {
         return changed;
     }
 
-    static class InternalAsyncFetch<T extends NodeOperationResponse> extends AsyncShardFetch<T> {
+    private final AtomicBoolean rerouting = new AtomicBoolean();
 
-        private final ClusterService clusterService;
-        private final AllocationService allocationService;
+    class InternalAsyncFetch<T extends NodeOperationResponse> extends AsyncShardFetch<T> {
 
-        public InternalAsyncFetch(ESLogger logger, String type, ShardId shardId, List<? extends NodesOperationResponse<T>, T> action,
-                                  ClusterService clusterService, AllocationService allocationService) {
+        public InternalAsyncFetch(ESLogger logger, String type, ShardId shardId, List<? extends NodesOperationResponse<T>, T> action) {
             super(logger, type, shardId, action);
-            this.clusterService = clusterService;
-            this.allocationService = allocationService;
         }
 
         @Override
         protected void reroute(ShardId shardId, String reason) {
-            clusterService.submitStateUpdateTask("async_shard_fetch(" + type + ") " + shardId + ", reasons (" + reason + ")", Priority.HIGH, new ClusterStateUpdateTask() {
+            if (rerouting.compareAndSet(false, true) == false) {
+                logger.trace("{} already has pending reroute, ignoring {}", shardId, reason);
+                return;
+            }
+            clusterService.submitStateUpdateTask("async_shard_fetch", Priority.HIGH, new ClusterStateUpdateTask() {
                 @Override
                 public ClusterState execute(ClusterState currentState) throws Exception {
+                    rerouting.set(false);
                     if (currentState.nodes().masterNode() == null) {
                         return currentState;
                     }
