Skip to content

Commit ffcec0e

Browse files
dnhatndnhatn
committed
Handle index deletion while querying in ES|QL
1 parent afd3a42 commit ffcec0e

File tree

2 files changed

+118
-21
lines changed

2 files changed

+118
-21
lines changed

‎x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/EsqlRetryIT.java‎

Lines changed: 69 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,9 @@
88
package org.elasticsearch.xpack.esql.action;
99

1010
import org.elasticsearch.action.index.IndexRequestBuilder;
11+
import org.elasticsearch.cluster.ClusterState;
12+
import org.elasticsearch.cluster.metadata.IndexMetadata;
13+
import org.elasticsearch.cluster.service.ClusterService;
1114
import org.elasticsearch.index.IndexService;
1215
import org.elasticsearch.index.shard.IndexShard;
1316
import org.elasticsearch.indices.IndicesService;
@@ -20,11 +23,14 @@
2023
import java.util.ArrayList;
2124
import java.util.Collection;
2225
import java.util.List;
26+
import java.util.concurrent.CountDownLatch;
27+
import java.util.concurrent.TimeUnit;
2328
import java.util.concurrent.atomic.AtomicBoolean;
2429

2530
import static org.elasticsearch.index.shard.IndexShardTestCase.closeShardNoCheck;
2631
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
2732
import static org.hamcrest.Matchers.equalTo;
33+
import static org.hamcrest.Matchers.not;
2834

2935
public class EsqlRetryIT extends AbstractEsqlIntegTestCase {
3036

@@ -59,6 +65,37 @@ public void testRetryOnShardFailures() throws Exception {
5965
}
6066
}
6167

68+
public void testQueryWhileDeletingIndices() {
69+
populateIndices();
70+
CountDownLatch waitForDeletion = new CountDownLatch(1);
71+
try {
72+
final AtomicBoolean deleted = new AtomicBoolean();
73+
for (String node : internalCluster().getNodeNames()) {
74+
MockTransportService.getInstance(node)
75+
.addRequestHandlingBehavior(ComputeService.DATA_ACTION_NAME, (handler, request, channel, task) -> {
76+
if (deleted.compareAndSet(false, true)) {
77+
deleteIndexCompletely("log-index-2");
78+
waitForDeletion.countDown();
79+
} else {
80+
assertTrue(waitForDeletion.await(10, TimeUnit.SECONDS));
81+
}
82+
handler.messageReceived(request, channel, task);
83+
});
84+
}
85+
EsqlQueryRequest request = new EsqlQueryRequest();
86+
request.query("FROM log-* | STATS COUNT(timestamp) | LIMIT 1");
87+
request.allowPartialResults(true);
88+
try (var resp = run(request)) {
89+
assertTrue(resp.isPartial());
90+
assertThat(EsqlTestUtils.getValuesList(resp).getFirst().getFirst(), equalTo(4L));
91+
}
92+
} finally {
93+
for (String node : internalCluster().getNodeNames()) {
94+
MockTransportService.getInstance(node).clearAllRules();
95+
}
96+
}
97+
}
98+
6299
private void populateIndices() {
63100
internalCluster().ensureAtLeastNumDataNodes(2);
64101
assertAcked(prepareCreate("log-index-1").setSettings(indexSettings(between(1, 3), 1)).setMapping("timestamp", "type=date"));
@@ -88,4 +125,36 @@ private void closeOrFailShards(String nodeName) throws Exception {
88125
}
89126
}
90127
}
128+
129+
/**
130+
* Deletes the given index and ensures it is completely removed from the cluster state and from all nodes
131+
*/
132+
private void deleteIndexCompletely(String indexName) throws Exception {
133+
assertAcked(indicesAdmin().prepareDelete(indexName));
134+
String[] nodeNames = internalCluster().getNodeNames();
135+
assertBusy(() -> {
136+
for (String nodeName : nodeNames) {
137+
ClusterState clusterState = internalCluster().getInstance(ClusterService.class, nodeName).state();
138+
for (IndexMetadata imd : clusterState.metadata().indicesAllProjects()) {
139+
assertThat(
140+
"Index [" + indexName + "] still exists in the cluster state on [" + nodeName + "]",
141+
imd.getIndex().getName(),
142+
not(equalTo(indexName))
143+
);
144+
}
145+
}
146+
for (String nodeName : nodeNames) {
147+
final IndicesService indicesService = internalCluster().getInstance(IndicesService.class, nodeName);
148+
for (IndexService indexService : indicesService) {
149+
for (IndexShard indexShard : indexService) {
150+
assertThat(
151+
"Index [" + indexName + "] still exists on node [" + nodeName + "]",
152+
indexShard.shardId().getIndexName(),
153+
not(equalTo(indexName))
154+
);
155+
}
156+
}
157+
}
158+
});
159+
}
91160
}

‎x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/DataNodeRequestSender.java‎

Lines changed: 49 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -27,8 +27,12 @@
2727
import org.elasticsearch.compute.operator.DriverCompletionInfo;
2828
import org.elasticsearch.compute.operator.FailureCollector;
2929
import org.elasticsearch.index.Index;
30+
import org.elasticsearch.index.IndexNotFoundException;
3031
import org.elasticsearch.index.query.QueryBuilder;
3132
import org.elasticsearch.index.shard.ShardId;
33+
import org.elasticsearch.index.shard.ShardNotFoundException;
34+
import org.elasticsearch.logging.LogManager;
35+
import org.elasticsearch.logging.Logger;
3236
import org.elasticsearch.search.SearchShardTarget;
3337
import org.elasticsearch.search.internal.AliasFilter;
3438
import org.elasticsearch.tasks.CancellableTask;
@@ -66,6 +70,8 @@
6670
*/
6771
abstract class DataNodeRequestSender {
6872

73+
private static final Logger LOGGER = LogManager.getLogger(DataNodeRequestSender.class);
74+
6975
/**
7076
* Query order according to the
7177
* <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/node-roles-overview.html">node roles</a>.
@@ -282,38 +288,51 @@ private void sendOneNodeRequest(TargetShards targetShards, ComputeListener compu
282288
final ActionListener<DriverCompletionInfo> listener = computeListener.acquireCompute();
283289
sendRequest(request.node, request.shardIds, request.aliasFilters, new NodeListener() {
284290

285-
void onAfter(DriverCompletionInfo info) {
291+
void onAfterRequest() {
286292
nodePermits.get(request.node).release();
287293
if (concurrentRequests != null) {
288294
concurrentRequests.release();
289295
}
290296
trySendingRequestsForPendingShards(targetShards, computeListener);
291-
listener.onResponse(info);
292297
}
293298

294299
@Override
295300
public void onResponse(DataNodeComputeResponse response) {
296-
// remove failures of successful shards
297-
for (ShardId shardId : request.shardIds()) {
298-
if (response.shardLevelFailures().containsKey(shardId) == false) {
299-
shardFailures.remove(shardId);
301+
try {
302+
// remove failures of successful shards
303+
for (ShardId shardId : request.shardIds()) {
304+
if (response.shardLevelFailures().containsKey(shardId) == false) {
305+
shardFailures.remove(shardId);
306+
}
300307
}
308+
for (var entry : response.shardLevelFailures().entrySet()) {
309+
final ShardId shardId = entry.getKey();
310+
trackShardLevelFailure(shardId, false, entry.getValue());
311+
pendingShardIds.add(shardId);
312+
}
313+
onAfterRequest();
314+
} catch (Exception ex) {
315+
expectNoFailure("expect no failure while handling data node response", ex);
316+
listener.onFailure(ex);
317+
return;
301318
}
302-
for (var entry : response.shardLevelFailures().entrySet()) {
303-
final ShardId shardId = entry.getKey();
304-
trackShardLevelFailure(shardId, false, entry.getValue());
305-
pendingShardIds.add(shardId);
306-
}
307-
onAfter(response.completionInfo());
319+
listener.onResponse(response.completionInfo());
308320
}
309321

310322
@Override
311323
public void onFailure(Exception e, boolean receivedData) {
312-
for (ShardId shardId : request.shardIds) {
313-
trackShardLevelFailure(shardId, receivedData, e);
314-
pendingShardIds.add(shardId);
324+
try {
325+
for (ShardId shardId : request.shardIds) {
326+
trackShardLevelFailure(shardId, receivedData, e);
327+
pendingShardIds.add(shardId);
328+
}
329+
onAfterRequest();
330+
} catch (Exception ex) {
331+
expectNoFailure("expect no failure while handling failure of data node request", ex);
332+
listener.onFailure(ex);
333+
return;
315334
}
316-
onAfter(DriverCompletionInfo.EMPTY);
335+
listener.onResponse(DriverCompletionInfo.EMPTY);
317336
}
318337

319338
@Override
@@ -325,6 +344,11 @@ public void onSkip() {
325344
onResponse(new DataNodeComputeResponse(DriverCompletionInfo.EMPTY, Map.of()));
326345
}
327346
}
347+
348+
private void expectNoFailure(String message, Exception e) {
349+
LOGGER.error(message, e);
350+
assert false : new AssertionError(message, e);
351+
}
328352
});
329353
}
330354

@@ -515,15 +539,19 @@ Map<ShardId, List<DiscoveryNode>> resolveShards(Set<ShardId> shardIds) {
515539
var project = projectResolver.getProjectState(clusterService.state());
516540
var nodes = Maps.<ShardId, List<DiscoveryNode>>newMapWithExpectedSize(shardIds.size());
517541
for (var shardId : shardIds) {
518-
nodes.put(
519-
shardId,
520-
project.routingTable()
542+
List<DiscoveryNode> allocatedNodes;
543+
try {
544+
allocatedNodes = project.routingTable()
521545
.shardRoutingTable(shardId)
522546
.allShards()
523547
.filter(shard -> shard.active() && shard.isSearchable())
524548
.map(shard -> project.cluster().nodes().get(shard.currentNodeId()))
525-
.toList()
526-
);
549+
.toList();
550+
} catch (Exception ex) {
551+
assert ex instanceof IndexNotFoundException || ex instanceof ShardNotFoundException : new AssertionError(ex);
552+
continue;
553+
}
554+
nodes.put(shardId, allocatedNodes);
527555
}
528556
return nodes;
529557
}

0 commit comments

Comments
 (0)