Skip to content

Commit a7f5874

Browse files
fcofdezfcofdez
authored
Check for global blocks after IndexNotFoundException in TransportMasterNodeAction (#78549)
Today we try to resolve index patterns to check cluster blocks in certain TransportMasterNodeActions, in some scenarios where a master node is recovering we could end up throwing a false IndexNotFoundException. This commit adds an extra check for global blocks when a IndexNotFoundException is thrown to ensure that we cover that case. Closes #70572 Backport of #78128
1 parent 1800ff3 commit a7f5874

File tree

2 files changed

+118
-4
lines changed

2 files changed

+118
-4
lines changed

‎server/src/main/java/org/elasticsearch/action/support/master/TransportMasterNodeAction.java‎

Lines changed: 21 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,9 @@
2121
import org.elasticsearch.cluster.ClusterStateObserver;
2222
import org.elasticsearch.cluster.MasterNodeChangePredicate;
2323
import org.elasticsearch.cluster.NotMasterException;
24+
import org.elasticsearch.cluster.block.ClusterBlock;
2425
import org.elasticsearch.cluster.block.ClusterBlockException;
26+
import org.elasticsearch.cluster.block.ClusterBlockLevel;
2527
import org.elasticsearch.cluster.coordination.FailedToCommitClusterStateException;
2628
import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
2729
import org.elasticsearch.cluster.node.DiscoveryNode;
@@ -30,6 +32,8 @@
3032
import org.elasticsearch.common.io.stream.Writeable;
3133
import org.elasticsearch.core.TimeValue;
3234
import org.elasticsearch.discovery.MasterNotDiscoveredException;
35+
import org.elasticsearch.gateway.GatewayService;
36+
import org.elasticsearch.index.IndexNotFoundException;
3337
import org.elasticsearch.node.NodeClosedException;
3438
import org.elasticsearch.tasks.CancellableTask;
3539
import org.elasticsearch.tasks.Task;
@@ -105,6 +109,21 @@ protected boolean localExecute(Request request) {
105109

106110
protected abstract ClusterBlockException checkBlock(Request request, ClusterState state);
107111

112+
private ClusterBlockException checkBlockIfStateRecovered(Request request, ClusterState state) {
113+
try {
114+
return checkBlock(request, state);
115+
} catch (IndexNotFoundException e) {
116+
if (state.blocks().hasGlobalBlock(GatewayService.STATE_NOT_RECOVERED_BLOCK)) {
117+
// no index metadata is exposed yet, but checkBlock depends on an index, so keep trying until the cluster forms
118+
assert GatewayService.STATE_NOT_RECOVERED_BLOCK.contains(ClusterBlockLevel.METADATA_READ);
119+
assert state.blocks().global(ClusterBlockLevel.METADATA_READ).stream().allMatch(ClusterBlock::retryable);
120+
return state.blocks().globalBlockedException(ClusterBlockLevel.METADATA_READ);
121+
} else {
122+
throw e;
123+
}
124+
}
125+
}
126+
108127
@Override
109128
protected void doExecute(Task task, final Request request, ActionListener<Response> listener) {
110129
ClusterState state = clusterService.state();
@@ -139,15 +158,15 @@ protected void doStart(ClusterState clusterState) {
139158
final DiscoveryNodes nodes = clusterState.nodes();
140159
if (nodes.isLocalNodeElectedMaster() || localExecute(request)) {
141160
// check for block, if blocked, retry, else, execute locally
142-
final ClusterBlockException blockException = checkBlock(request, clusterState);
161+
final ClusterBlockException blockException = checkBlockIfStateRecovered(request, clusterState);
143162
if (blockException != null) {
144163
if (blockException.retryable() == false) {
145164
listener.onFailure(blockException);
146165
} else {
147166
logger.debug("can't execute due to a cluster block, retrying", blockException);
148167
retry(clusterState, blockException, newState -> {
149168
try {
150-
ClusterBlockException newException = checkBlock(request, newState);
169+
ClusterBlockException newException = checkBlockIfStateRecovered(request, newState);
151170
return (newException == null || newException.retryable() == false);
152171
} catch (Exception e) {
153172
// accept state as block will be rechecked by doStart() and listener.onFailure() then called

‎server/src/test/java/org/elasticsearch/action/support/master/TransportMasterNodeActionTests.java‎

Lines changed: 97 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -13,8 +13,10 @@
1313
import org.elasticsearch.action.ActionListener;
1414
import org.elasticsearch.action.ActionRequestValidationException;
1515
import org.elasticsearch.action.ActionResponse;
16+
import org.elasticsearch.action.IndicesRequest;
1617
import org.elasticsearch.action.support.ActionFilters;
1718
import org.elasticsearch.action.support.ActionTestUtils;
19+
import org.elasticsearch.action.support.IndicesOptions;
1820
import org.elasticsearch.action.support.PlainActionFuture;
1921
import org.elasticsearch.action.support.ThreadedActionListener;
2022
import org.elasticsearch.action.support.replication.ClusterStateCreationUtils;
@@ -25,15 +27,19 @@
2527
import org.elasticsearch.cluster.block.ClusterBlockLevel;
2628
import org.elasticsearch.cluster.block.ClusterBlocks;
2729
import org.elasticsearch.cluster.coordination.FailedToCommitClusterStateException;
30+
import org.elasticsearch.cluster.metadata.IndexMetadata;
31+
import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
32+
import org.elasticsearch.cluster.metadata.Metadata;
2833
import org.elasticsearch.cluster.node.DiscoveryNode;
2934
import org.elasticsearch.cluster.node.DiscoveryNodeRole;
3035
import org.elasticsearch.cluster.node.DiscoveryNodes;
3136
import org.elasticsearch.cluster.service.ClusterService;
37+
import org.elasticsearch.common.Strings;
3238
import org.elasticsearch.common.io.stream.StreamInput;
3339
import org.elasticsearch.common.io.stream.StreamOutput;
3440
import org.elasticsearch.common.settings.Settings;
35-
import org.elasticsearch.core.TimeValue;
3641
import org.elasticsearch.common.util.concurrent.EsThreadPoolExecutor;
42+
import org.elasticsearch.core.TimeValue;
3743
import org.elasticsearch.discovery.MasterNotDiscoveredException;
3844
import org.elasticsearch.indices.TestIndexNameExpressionResolver;
3945
import org.elasticsearch.node.NodeClosedException;
@@ -65,6 +71,7 @@
6571
import java.util.concurrent.ExecutionException;
6672
import java.util.concurrent.TimeUnit;
6773

74+
import static org.elasticsearch.gateway.GatewayService.STATE_NOT_RECOVERED_BLOCK;
6875
import static org.elasticsearch.test.ClusterServiceUtils.createClusterService;
6976
import static org.elasticsearch.test.ClusterServiceUtils.setState;
7077
import static org.hamcrest.Matchers.equalTo;
@@ -124,7 +131,9 @@ void assertListenerThrows(String msg, ActionFuture<?> listener, Class<?> klass)
124131
}
125132
}
126133

127-
public static class Request extends MasterNodeRequest<Request> {
134+
public static class Request extends MasterNodeRequest<Request> implements IndicesRequest.Replaceable {
135+
private String[] indices = Strings.EMPTY_ARRAY;
136+
128137
Request() {}
129138

130139
Request(StreamInput in) throws IOException {
@@ -140,6 +149,22 @@ public ActionRequestValidationException validate() {
140149
public Task createTask(long id, String type, String action, TaskId parentTaskId, Map<String, String> headers) {
141150
return new CancellableTask(id, type, action, "", parentTaskId, headers);
142151
}
152+
153+
@Override
154+
public String[] indices() {
155+
return indices;
156+
}
157+
158+
@Override
159+
public IndicesOptions indicesOptions() {
160+
return IndicesOptions.strictExpandOpen();
161+
}
162+
163+
@Override
164+
public IndicesRequest indices(String... indices) {
165+
this.indices = indices;
166+
return this;
167+
}
143168
}
144169

145170
class Response extends ActionResponse {
@@ -567,6 +592,76 @@ public void testTaskCancellationOnceActionItIsDispatchedToMaster() throws Except
567592
expectThrows(CancellationException.class, listener::actionGet);
568593
}
569594

595+
public void testGlobalBlocksAreCheckedAfterIndexNotFoundException() throws Exception {
596+
Request request = new Request().masterNodeTimeout(TimeValue.timeValueSeconds(60));
597+
String indexRequestName = "my-index";
598+
request.indices(indexRequestName);
599+
600+
ClusterState stateWithBlockWithoutIndexMetadata =
601+
ClusterState.builder(ClusterStateCreationUtils.state(localNode, localNode, allNodes))
602+
.blocks(ClusterBlocks.builder().addGlobalBlock(STATE_NOT_RECOVERED_BLOCK))
603+
.build();
604+
setState(clusterService, stateWithBlockWithoutIndexMetadata);
605+
606+
Action action = new Action("internal:testAction", transportService, clusterService, threadPool, ThreadPool.Names.SAME) {
607+
final IndexNameExpressionResolver indexNameExpressionResolver = TestIndexNameExpressionResolver.newInstance();
608+
609+
@Override
610+
protected ClusterBlockException checkBlock(Request request, ClusterState state) {
611+
return state.blocks().indicesBlockedException(ClusterBlockLevel.METADATA_READ,
612+
indexNameExpressionResolver.concreteIndexNamesWithSystemIndexAccess(state, request));
613+
}
614+
615+
};
616+
617+
PlainActionFuture<Response> listener = new PlainActionFuture<>();
618+
ActionTestUtils.execute(action, null, request, listener);
619+
620+
assertFalse(listener.isDone());
621+
IndexMetadata.Builder indexMetadataBuilder =
622+
IndexMetadata.builder(indexRequestName)
623+
.settings(settings(Version.CURRENT))
624+
.numberOfShards(1)
625+
.numberOfReplicas(0);
626+
ClusterState clusterStateWithoutBlocks = ClusterState.builder(ClusterStateCreationUtils.state(localNode, localNode, allNodes))
627+
.metadata(Metadata.builder().put(indexMetadataBuilder).build())
628+
.blocks(ClusterBlocks.EMPTY_CLUSTER_BLOCK)
629+
.build();
630+
setState(clusterService, clusterStateWithoutBlocks);
631+
assertTrue(listener.isDone());
632+
listener.get();
633+
}
634+
635+
public void testGlobalBlocksAreCheckedAfterIndexNotFoundExceptionTimesOutIfIndexIsNotFound() {
636+
Request request = new Request().masterNodeTimeout(TimeValue.timeValueMillis(50));
637+
String indexRequestName = "my-index";
638+
request.indices(indexRequestName);
639+
640+
ClusterState stateWithBlockWithoutIndexMetadata =
641+
ClusterState.builder(ClusterStateCreationUtils.state(localNode, localNode, allNodes))
642+
.blocks(ClusterBlocks.builder().addGlobalBlock(STATE_NOT_RECOVERED_BLOCK))
643+
.build();
644+
setState(clusterService, stateWithBlockWithoutIndexMetadata);
645+
646+
Action action = new Action("internal:testAction", transportService, clusterService, threadPool, ThreadPool.Names.SAME) {
647+
final IndexNameExpressionResolver indexNameExpressionResolver = TestIndexNameExpressionResolver.newInstance();
648+
649+
@Override
650+
protected ClusterBlockException checkBlock(Request request, ClusterState state) {
651+
return state.blocks().indicesBlockedException(ClusterBlockLevel.METADATA_READ,
652+
indexNameExpressionResolver.concreteIndexNamesWithSystemIndexAccess(state, request));
653+
}
654+
655+
};
656+
657+
PlainActionFuture<Response> listener = new PlainActionFuture<>();
658+
ActionTestUtils.execute(action, null, request, listener);
659+
660+
ExecutionException ex = expectThrows(ExecutionException.class, listener::get);
661+
assertThat(ex.getCause(), instanceOf(MasterNotDiscoveredException.class));
662+
assertThat(ex.getCause().getCause(), instanceOf(ClusterBlockException.class));
663+
}
664+
570665
private Runnable blockAllThreads(String executorName) throws Exception {
571666
final int numberOfThreads = threadPool.info(executorName).getMax();
572667
final EsThreadPoolExecutor executor = (EsThreadPoolExecutor) threadPool.executor(executorName);

0 commit comments

Comments
 (0)