From 9149630749ea76d1294f8c3409f57a76a1af6d05 Mon Sep 17 00:00:00 2001
From: Jason Tedor <jason@tedor.me>
Date: Wed, 16 Dec 2015 18:45:22 -0500
Subject: [PATCH] Reroute once per batch of shard failures

This commit modifies the behavior after publication of a new cluster
state to only invoke the reroute logic once per batch of shard failures
rather than once per shard failure.
---
 .../action/shard/ShardStateAction.java        | 29 ++++++++++---------
 1 file changed, 15 insertions(+), 14 deletions(-)

diff --git a/core/src/main/java/org/elasticsearch/cluster/action/shard/ShardStateAction.java b/core/src/main/java/org/elasticsearch/cluster/action/shard/ShardStateAction.java
index 3a01ced6ebf..a04a6d7bd51 100644
--- a/core/src/main/java/org/elasticsearch/cluster/action/shard/ShardStateAction.java
+++ b/core/src/main/java/org/elasticsearch/cluster/action/shard/ShardStateAction.java
@@ -148,20 +148,9 @@ public class ShardStateAction extends AbstractComponent {
                     @Override
                     public void clusterStateProcessed(String source, ClusterState oldState, ClusterState newState) {
                         try {
-                            int numberOfUnassignedShards = newState.getRoutingNodes().unassigned().size();
-                            if (oldState != newState && numberOfUnassignedShards > 0) {
-                                String reason = String.format(Locale.ROOT, "[%d] unassigned shards after failing shard [%s]", numberOfUnassignedShards, request.shardRouting);
-                                if (logger.isTraceEnabled()) {
-                                    logger.trace(reason + ", scheduling a reroute");
-                                }
-                                routingService.reroute(reason);
-                            }
-                        } finally {
-                            try {
-                                channel.sendResponse(TransportResponse.Empty.INSTANCE);
-                            } catch (Throwable channelThrowable) {
-                                logger.warn("failed to send response while failing shard [{}]", channelThrowable, request.shardRouting);
-                            }
+                            channel.sendResponse(TransportResponse.Empty.INSTANCE);
+                        } catch (Throwable channelThrowable) {
+                            logger.warn("failed to send response while failing shard [{}]", channelThrowable, request.shardRouting);
                         }
                     }
                 }
@@ -189,6 +178,18 @@ public class ShardStateAction extends AbstractComponent {
             }
             return batchResultBuilder.build(maybeUpdatedState);
         }
+
+        @Override
+        public void clusterStatePublished(ClusterState newClusterState) {
+            int numberOfUnassignedShards = newClusterState.getRoutingNodes().unassigned().size();
+            if (numberOfUnassignedShards > 0) {
+                String reason = String.format(Locale.ROOT, "[%d] unassigned shards after failing shards", numberOfUnassignedShards);
+                if (logger.isTraceEnabled()) {
+                    logger.trace(reason + ", scheduling a reroute");
+                }
+                routingService.reroute(reason);
+            }
+        }
     }
 
     private final ShardFailedClusterStateHandler shardFailedClusterStateHandler = new ShardFailedClusterStateHandler();