From 72682a6730d051709cb274aff53b124d5bf682f9 Mon Sep 17 00:00:00 2001
From: kimchy <kimchy@gmail.com>
Date: Fri, 23 Jul 2010 01:59:56 +0300
Subject: [PATCH] always retry on connect exception with master operations

---
 .../TransportMasterNodeOperationAction.java   | 64 +++++++++----------
 .../IndexLifecycleActionTests.java            |  3 +
 2 files changed, 33 insertions(+), 34 deletions(-)

diff --git a/modules/elasticsearch/src/main/java/org/elasticsearch/action/support/master/TransportMasterNodeOperationAction.java b/modules/elasticsearch/src/main/java/org/elasticsearch/action/support/master/TransportMasterNodeOperationAction.java
index 52ff4b29075..cac95e2bc91 100644
--- a/modules/elasticsearch/src/main/java/org/elasticsearch/action/support/master/TransportMasterNodeOperationAction.java
+++ b/modules/elasticsearch/src/main/java/org/elasticsearch/action/support/master/TransportMasterNodeOperationAction.java
@@ -138,41 +138,37 @@ public abstract class TransportMasterNodeOperationAction<Request extends MasterN
                 }
 
                 @Override public void handleException(final RemoteTransportException exp) {
-                    if (retrying) {
-                        listener.onFailure(exp);
+                    if (exp.unwrapCause() instanceof ConnectTransportException) {
+                        // we want to retry here a bit to see if a new master is elected
+                        clusterService.add(request.masterNodeTimeout(), new TimeoutClusterStateListener() {
+                            @Override public void postAdded() {
+                                ClusterState clusterStateV2 = clusterService.state();
+                                if (!clusterState.nodes().masterNodeId().equals(clusterStateV2.nodes().masterNodeId())) {
+                                    // master changes while adding the listener, try here
+                                    clusterService.remove(this);
+                                    innerExecute(request, listener, false);
+                                }
+                            }
+
+                            @Override public void onClose() {
+                                clusterService.remove(this);
+                                listener.onFailure(new NodeClosedException(nodes.localNode()));
+                            }
+
+                            @Override public void onTimeout(TimeValue timeout) {
+                                clusterService.remove(this);
+                                listener.onFailure(exp);
+                            }
+
+                            @Override public void clusterChanged(ClusterChangedEvent event) {
+                                if (event.nodesDelta().masterNodeChanged()) {
+                                    clusterService.remove(this);
+                                    innerExecute(request, listener, false);
+                                }
+                            }
+                        });
                     } else {
-                        if (exp.unwrapCause() instanceof ConnectTransportException) {
-                            // we want to retry here a bit to see if a new master is elected
-                            clusterService.add(request.masterNodeTimeout(), new TimeoutClusterStateListener() {
-                                @Override public void postAdded() {
-                                    ClusterState clusterStateV2 = clusterService.state();
-                                    if (!clusterState.nodes().masterNodeId().equals(clusterStateV2.nodes().masterNodeId())) {
-                                        // master changes while adding the listener, try here 
-                                        clusterService.remove(this);
-                                        innerExecute(request, listener, true);
-                                    }
-                                }
-
-                                @Override public void onClose() {
-                                    clusterService.remove(this);
-                                    listener.onFailure(new NodeClosedException(nodes.localNode()));
-                                }
-
-                                @Override public void onTimeout(TimeValue timeout) {
-                                    clusterService.remove(this);
-                                    listener.onFailure(exp);
-                                }
-
-                                @Override public void clusterChanged(ClusterChangedEvent event) {
-                                    if (event.nodesDelta().masterNodeChanged()) {
-                                        clusterService.remove(this);
-                                        innerExecute(request, listener, true);
-                                    }
-                                }
-                            });
-                        } else {
-                            listener.onFailure(exp);
-                        }
+                        listener.onFailure(exp);
                     }
                 }
             });
diff --git a/modules/test/integration/src/test/java/org/elasticsearch/test/integration/indexlifecycle/IndexLifecycleActionTests.java b/modules/test/integration/src/test/java/org/elasticsearch/test/integration/indexlifecycle/IndexLifecycleActionTests.java
index 76b80ae7e68..19d173f2334 100644
--- a/modules/test/integration/src/test/java/org/elasticsearch/test/integration/indexlifecycle/IndexLifecycleActionTests.java
+++ b/modules/test/integration/src/test/java/org/elasticsearch/test/integration/indexlifecycle/IndexLifecycleActionTests.java
@@ -111,6 +111,7 @@ public class IndexLifecycleActionTests extends AbstractNodesTests {
         logger.info("Starting server3");
         // start another server
         startNode("server3", settings);
+        Thread.sleep(200);
 
         ClusterService clusterService3 = ((InternalNode) node("server3")).injector().getInstance(ClusterService.class);
 
@@ -214,6 +215,7 @@ public class IndexLifecycleActionTests extends AbstractNodesTests {
         // start another server
         logger.info("Starting server2");
         startNode("server2", settings);
+        Thread.sleep(200);
 
         logger.info("Running Cluster Health");
         clusterHealth = client("server1").admin().cluster().health(clusterHealth().waitForGreenStatus().waitForRelocatingShards(0).waitForNodes("2")).actionGet();
@@ -240,6 +242,7 @@ public class IndexLifecycleActionTests extends AbstractNodesTests {
         // start another server
         logger.info("Starting server3");
         startNode("server3");
+        Thread.sleep(200);
 
         ClusterService clusterService3 = ((InternalNode) node("server3")).injector().getInstance(ClusterService.class);