|
@@ -145,69 +145,79 @@ public abstract class TransportMasterNodeAction<Request extends MasterNodeReques
|
|
|
}
|
|
|
|
|
|
protected void doStart(ClusterState clusterState) {
|
|
|
- final Predicate<ClusterState> masterChangePredicate = MasterNodeChangePredicate.build(clusterState);
|
|
|
- final DiscoveryNodes nodes = clusterState.nodes();
|
|
|
- if (nodes.isLocalNodeElectedMaster() || localExecute(request)) {
|
|
|
- // check for block, if blocked, retry, else, execute locally
|
|
|
- final ClusterBlockException blockException = checkBlock(request, clusterState);
|
|
|
- if (blockException != null) {
|
|
|
- if (!blockException.retryable()) {
|
|
|
- listener.onFailure(blockException);
|
|
|
- } else {
|
|
|
- logger.trace("can't execute due to a cluster block, retrying", blockException);
|
|
|
- retry(blockException, newState -> {
|
|
|
- ClusterBlockException newException = checkBlock(request, newState);
|
|
|
- return (newException == null || !newException.retryable());
|
|
|
- });
|
|
|
- }
|
|
|
- } else {
|
|
|
- ActionListener<Response> delegate = new ActionListener<Response>() {
|
|
|
- @Override
|
|
|
- public void onResponse(Response response) {
|
|
|
- listener.onResponse(response);
|
|
|
+ try {
|
|
|
+ final Predicate<ClusterState> masterChangePredicate = MasterNodeChangePredicate.build(clusterState);
|
|
|
+ final DiscoveryNodes nodes = clusterState.nodes();
|
|
|
+ if (nodes.isLocalNodeElectedMaster() || localExecute(request)) {
|
|
|
+ // check for block, if blocked, retry, else, execute locally
|
|
|
+ final ClusterBlockException blockException = checkBlock(request, clusterState);
|
|
|
+ if (blockException != null) {
|
|
|
+ if (!blockException.retryable()) {
|
|
|
+ listener.onFailure(blockException);
|
|
|
+ } else {
|
|
|
+ logger.trace("can't execute due to a cluster block, retrying", blockException);
|
|
|
+ retry(blockException, newState -> {
|
|
|
+ try {
|
|
|
+ ClusterBlockException newException = checkBlock(request, newState);
|
|
|
+ return (newException == null || !newException.retryable());
|
|
|
+ } catch (Exception e) {
|
|
|
+ // accept state as block will be rechecked by doStart() and listener.onFailure() then called
|
|
|
+ logger.trace("exception occurred during cluster block checking, accepting state", e);
|
|
|
+ return true;
|
|
|
+ }
|
|
|
+ });
|
|
|
}
|
|
|
+ } else {
|
|
|
+ ActionListener<Response> delegate = new ActionListener<Response>() {
|
|
|
+ @Override
|
|
|
+ public void onResponse(Response response) {
|
|
|
+ listener.onResponse(response);
|
|
|
+ }
|
|
|
|
|
|
- @Override
|
|
|
- public void onFailure(Exception t) {
|
|
|
- if (t instanceof Discovery.FailedToCommitClusterStateException
|
|
|
+ @Override
|
|
|
+ public void onFailure(Exception t) {
|
|
|
+ if (t instanceof Discovery.FailedToCommitClusterStateException
|
|
|
|| (t instanceof NotMasterException)) {
|
|
|
- logger.debug(() -> new ParameterizedMessage("master could not publish cluster state or stepped down before publishing action [{}], scheduling a retry", actionName), t);
|
|
|
- retry(t, masterChangePredicate);
|
|
|
- } else {
|
|
|
- listener.onFailure(t);
|
|
|
+ logger.debug(() -> new ParameterizedMessage("master could not publish cluster state or stepped down before publishing action [{}], scheduling a retry", actionName), t);
|
|
|
+ retry(t, masterChangePredicate);
|
|
|
+ } else {
|
|
|
+ listener.onFailure(t);
|
|
|
+ }
|
|
|
}
|
|
|
- }
|
|
|
- };
|
|
|
- threadPool.executor(executor).execute(new ActionRunnable(delegate) {
|
|
|
- @Override
|
|
|
- protected void doRun() throws Exception {
|
|
|
- masterOperation(task, request, clusterState, delegate);
|
|
|
- }
|
|
|
- });
|
|
|
- }
|
|
|
- } else {
|
|
|
- if (nodes.getMasterNode() == null) {
|
|
|
- logger.debug("no known master node, scheduling a retry");
|
|
|
- retry(null, masterChangePredicate);
|
|
|
+ };
|
|
|
+ threadPool.executor(executor).execute(new ActionRunnable(delegate) {
|
|
|
+ @Override
|
|
|
+ protected void doRun() throws Exception {
|
|
|
+ masterOperation(task, request, clusterState, delegate);
|
|
|
+ }
|
|
|
+ });
|
|
|
+ }
|
|
|
} else {
|
|
|
- DiscoveryNode masterNode = nodes.getMasterNode();
|
|
|
- final String actionName = getMasterActionName(masterNode);
|
|
|
- transportService.sendRequest(masterNode, actionName, request, new ActionListenerResponseHandler<Response>(listener,
|
|
|
- TransportMasterNodeAction.this::newResponse) {
|
|
|
- @Override
|
|
|
- public void handleException(final TransportException exp) {
|
|
|
- Throwable cause = exp.unwrapCause();
|
|
|
- if (cause instanceof ConnectTransportException) {
|
|
|
- // we want to retry here a bit to see if a new master is elected
|
|
|
- logger.debug("connection exception while trying to forward request with action name [{}] to master node [{}], scheduling a retry. Error: [{}]",
|
|
|
+ if (nodes.getMasterNode() == null) {
|
|
|
+ logger.debug("no known master node, scheduling a retry");
|
|
|
+ retry(null, masterChangePredicate);
|
|
|
+ } else {
|
|
|
+ DiscoveryNode masterNode = nodes.getMasterNode();
|
|
|
+ final String actionName = getMasterActionName(masterNode);
|
|
|
+ transportService.sendRequest(masterNode, actionName, request, new ActionListenerResponseHandler<Response>(listener,
|
|
|
+ TransportMasterNodeAction.this::newResponse) {
|
|
|
+ @Override
|
|
|
+ public void handleException(final TransportException exp) {
|
|
|
+ Throwable cause = exp.unwrapCause();
|
|
|
+ if (cause instanceof ConnectTransportException) {
|
|
|
+ // we want to retry here a bit to see if a new master is elected
|
|
|
+ logger.debug("connection exception while trying to forward request with action name [{}] to master node [{}], scheduling a retry. Error: [{}]",
|
|
|
actionName, nodes.getMasterNode(), exp.getDetailedMessage());
|
|
|
- retry(cause, masterChangePredicate);
|
|
|
- } else {
|
|
|
- listener.onFailure(exp);
|
|
|
+ retry(cause, masterChangePredicate);
|
|
|
+ } else {
|
|
|
+ listener.onFailure(exp);
|
|
|
+ }
|
|
|
}
|
|
|
- }
|
|
|
- });
|
|
|
+ });
|
|
|
+ }
|
|
|
}
|
|
|
+ } catch (Exception e) {
|
|
|
+ listener.onFailure(e);
|
|
|
}
|
|
|
}
|
|
|
|