honeymoose/OpenSearch
1
0
Fork 0
mirror of https://github.com/honeymoose/OpenSearch.git synced 2025-02-17 02:14:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

always retry on connect exception with master operations

Browse Source
This commit is contained in:
kimchy 2010-07-23 01:59:56 +03:00
parent 9283e2a7ad
commit 72682a6730
2 changed files with 33 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

64
modules/elasticsearch/src/main/java/org/elasticsearch/action/support/master/TransportMasterNodeOperationAction.java
View File

@ -138,41 +138,37 @@ public abstract class TransportMasterNodeOperationAction<Request extends MasterN
} }
@Override public void handleException(final RemoteTransportException exp) { @Override public void handleException(final RemoteTransportException exp) {
if (retrying) { if (exp.unwrapCause() instanceof ConnectTransportException) {
listener.onFailure(exp); // 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 { } else {
if (exp.unwrapCause() instanceof ConnectTransportException) { listener.onFailure(exp);
// 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);
}
} }
} }
}); });

3
modules/test/integration/src/test/java/org/elasticsearch/test/integration/indexlifecycle/IndexLifecycleActionTests.java
View File

@ -111,6 +111,7 @@ public class IndexLifecycleActionTests extends AbstractNodesTests {
logger.info("Starting server3"); logger.info("Starting server3");
// start another server // start another server
startNode("server3", settings); startNode("server3", settings);
Thread.sleep(200);
ClusterService clusterService3 = ((InternalNode) node("server3")).injector().getInstance(ClusterService.class); ClusterService clusterService3 = ((InternalNode) node("server3")).injector().getInstance(ClusterService.class);
@ -214,6 +215,7 @@ public class IndexLifecycleActionTests extends AbstractNodesTests {
// start another server // start another server
logger.info("Starting server2"); logger.info("Starting server2");
startNode("server2", settings); startNode("server2", settings);
Thread.sleep(200);
logger.info("Running Cluster Health"); logger.info("Running Cluster Health");
clusterHealth = client("server1").admin().cluster().health(clusterHealth().waitForGreenStatus().waitForRelocatingShards(0).waitForNodes("2")).actionGet(); 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 // start another server
logger.info("Starting server3"); logger.info("Starting server3");
startNode("server3"); startNode("server3");
Thread.sleep(200);
ClusterService clusterService3 = ((InternalNode) node("server3")).injector().getInstance(ClusterService.class); ClusterService clusterService3 = ((InternalNode) node("server3")).injector().getInstance(ClusterService.class);