Bulk process of shard started/failed should not execute on already pr…

…ocessed events closes #5061
elastic · Feb 10, 2014 · 83b09d2 · 83b09d2
1 parent 85b7604
commit 83b09d2
Showing 1 changed file with 11 additions and 0 deletions.
diff --git a/src/main/java/org/elasticsearch/cluster/action/shard/ShardStateAction.java b/src/main/java/org/elasticsearch/cluster/action/shard/ShardStateAction.java
@@ -119,6 +119,9 @@ private void innerShardFailed(final ShardRoutingEntry shardRoutingEntry) {
         clusterService.submitStateUpdateTask("shard-failed (" + shardRoutingEntry.shardRouting + "), reason [" + shardRoutingEntry.reason + "]", Priority.HIGH, new ClusterStateUpdateTask() {
             @Override
             public ClusterState execute(ClusterState currentState) {
+                if (shardRoutingEntry.processed) {
+                    return currentState;
+                }
 
                 List<ShardRoutingEntry> shardRoutingEntries = new ArrayList<ShardRoutingEntry>();
                 failedShardQueue.drainTo(shardRoutingEntries);
@@ -133,6 +136,7 @@ public ClusterState execute(ClusterState currentState) {
                 List<ShardRouting> shardRoutingsToBeApplied = new ArrayList<ShardRouting>(shardRoutingEntries.size());
                 for (int i = 0; i < shardRoutingEntries.size(); i++) {
                     ShardRoutingEntry shardRoutingEntry = shardRoutingEntries.get(i);
+                    shardRoutingEntry.processed = true;
                     ShardRouting shardRouting = shardRoutingEntry.shardRouting;
                     IndexMetaData indexMetaData = metaData.index(shardRouting.index());
                     // if there is no metadata or the current index is not of the right uuid, the index has been deleted while it was being allocated
@@ -176,6 +180,10 @@ private void innerShardStarted(final ShardRoutingEntry shardRoutingEntry) {
                     @Override
                     public ClusterState execute(ClusterState currentState) {
 
+                        if (shardRoutingEntry.processed) {
+                            return currentState;
+                        }
+
                         List<ShardRoutingEntry> shardRoutingEntries = new ArrayList<ShardRoutingEntry>();
                         startedShardsQueue.drainTo(shardRoutingEntries);
 
@@ -191,6 +199,7 @@ public ClusterState execute(ClusterState currentState) {
 
                         for (int i = 0; i < shardRoutingEntries.size(); i++) {
                             ShardRoutingEntry shardRoutingEntry = shardRoutingEntries.get(i);
+                            shardRoutingEntry.processed = true;
                             ShardRouting shardRouting = shardRoutingEntry.shardRouting;
                             try {
                                 IndexMetaData indexMetaData = metaData.index(shardRouting.index());
@@ -306,6 +315,8 @@ static class ShardRoutingEntry extends TransportRequest {
 
         private String reason;
 
+        volatile boolean processed; // state field, no need to serialize
+
         private ShardRoutingEntry() {
         }