Skip to content

Fix race condition when resolving new location for multiple shards at once #128062

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
May 20, 2025
Merged

Fix race condition when resolving new location for multiple shards at once #128062

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
5ea6506
test to demonstrate the issue
idegtiarenko May 14, 2025
cff6ddb
fix
idegtiarenko May 14, 2025
3b35bb5
Merge branch 'main' into fix_127188
idegtiarenko May 20, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 7 additions & 3 deletions .../plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/DataNodeRequestSender.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -192,8 +192,7 @@ private void trySendingRequestsForPendingShards(TargetShards targetShards, Compu
var pendingRetries = new HashSet<ShardId>();
for (ShardId shardId : pendingShardIds) {
if (targetShards.getShard(shardId).remainingNodes.isEmpty()) {
var failure = shardFailures.get(shardId);
if (failure != null && failure.fatal == false && failure.failure instanceof NoShardAvailableActionException) {
if (isRetryableFailure(shardFailures.get(shardId))) {
pendingRetries.add(shardId);
}
}
Expand All @@ -204,7 +203,8 @@ private void trySendingRequestsForPendingShards(TargetShards targetShards, Compu
}
}
for (ShardId shardId : pendingShardIds) {
if (targetShards.getShard(shardId).remainingNodes.isEmpty()) {
if (targetShards.getShard(shardId).remainingNodes.isEmpty()
&& (isRetryableFailure(shardFailures.get(shardId)) == false || pendingRetries.contains(shardId))) {
Copy link
Contributor Author

@idegtiarenko idegtiarenko May 14, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It was previously possible that between executing line 195 and line 206 one of the data nodes returns with NoSuchShard exception and add a new pendingShardId. As a result such shard might still not have remainingNodes nodes to query. This change supposed to detect such situations and delay such resolution to the next round.

This was detected by testSearchWhileRelocating integration test that is fairly slow and expensive so I added testRetryMultipleMovedShards unit test that made it much easier to reproduce.

shardFailures.compute(
shardId,
(k, v) -> new ShardFailure(
Expand Down Expand Up @@ -378,6 +378,10 @@ record NodeRequest(DiscoveryNode node, List<ShardId> shardIds, Map<Index, AliasF

private record ShardFailure(boolean fatal, Exception failure) {}

private static boolean isRetryableFailure(ShardFailure failure) {
return failure != null && failure.fatal == false && failure.failure instanceof NoShardAvailableActionException;
}

/**
* Selects the next nodes to send requests to. Limits to at most one outstanding request per node.
* If there is already a request in-flight to a node, another request will not be sent to the same node
Expand Down
60 changes: 41 additions & 19 deletions ...in/esql/src/test/java/org/elasticsearch/xpack/esql/plugin/DataNodeRequestSenderTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,8 +55,8 @@
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import java.util.stream.Collectors;

import static java.util.stream.Collectors.toMap;
import static org.elasticsearch.cluster.node.DiscoveryNodeRole.DATA_COLD_NODE_ROLE;
import static org.elasticsearch.cluster.node.DiscoveryNodeRole.DATA_FROZEN_NODE_ROLE;
import static org.elasticsearch.cluster.node.DiscoveryNodeRole.DATA_HOT_NODE_ROLE;
Expand Down Expand Up @@ -450,6 +450,32 @@ public void testRetryMovedShard() {
assertThat(attempt.get(), equalTo(3));
}

public void testRetryMultipleMovedShards() {
var attempt = new AtomicInteger(0);
var response = safeGet(
sendRequests(
randomBoolean(),
-1,
List.of(targetShard(shard1, node1), targetShard(shard2, node2), targetShard(shard3, node3)),
shardIds -> shardIds.stream().collect(toMap(Function.identity(), shardId -> List.of(randomFrom(node1, node2, node3)))),
(node, shardIds, aliasFilters, listener) -> runWithDelay(
() -> listener.onResponse(
attempt.incrementAndGet() <= 6
? new DataNodeComputeResponse(
DriverCompletionInfo.EMPTY,
shardIds.stream().collect(toMap(Function.identity(), ShardNotFoundException::new))
)
: new DataNodeComputeResponse(DriverCompletionInfo.EMPTY, Map.of())
)
)
)
);
assertThat(response.totalShards, equalTo(3));
assertThat(response.successfulShards, equalTo(3));
assertThat(response.skippedShards, equalTo(0));
assertThat(response.failedShards, equalTo(0));
}

public void testDoesNotRetryMovedShardIndefinitely() {
var attempt = new AtomicInteger(0);
var response = safeGet(sendRequests(true, -1, List.of(targetShard(shard1, node1)), shardIds -> {
Expand Down Expand Up @@ -517,28 +543,28 @@ public void testRetryUnassignedShardWithoutPartialResults() {

);
expectThrows(NoShardAvailableActionException.class, containsString("no such shard"), future::actionGet);
assertThat(attempt.get(), equalTo(1));
}

public void testRetryUnassignedShardWithPartialResults() {
var response = safeGet(
sendRequests(
true,
-1,
List.of(targetShard(shard1, node1), targetShard(shard2, node2)),
shardIds -> Map.of(shard1, List.of()),
(node, shardIds, aliasFilters, listener) -> runWithDelay(
() -> listener.onResponse(
Objects.equals(shardIds, List.of(shard2))
? new DataNodeComputeResponse(DriverCompletionInfo.EMPTY, Map.of())
: new DataNodeComputeResponse(DriverCompletionInfo.EMPTY, Map.of(shard1, new ShardNotFoundException(shard1)))
)
var attempt = new AtomicInteger(0);
var response = safeGet(sendRequests(true, -1, List.of(targetShard(shard1, node1), targetShard(shard2, node2)), shardIds -> {
attempt.incrementAndGet();
return Map.of(shard1, List.of());
},
(node, shardIds, aliasFilters, listener) -> runWithDelay(
() -> listener.onResponse(
Objects.equals(shardIds, List.of(shard2))
? new DataNodeComputeResponse(DriverCompletionInfo.EMPTY, Map.of())
: new DataNodeComputeResponse(DriverCompletionInfo.EMPTY, Map.of(shard1, new ShardNotFoundException(shard1)))
)
)
);
));
assertThat(response.totalShards, equalTo(2));
assertThat(response.successfulShards, equalTo(1));
assertThat(response.skippedShards, equalTo(0));
assertThat(response.failedShards, equalTo(1));
assertThat(attempt.get(), equalTo(1));
}

static DataNodeRequestSender.TargetShard targetShard(ShardId shardId, DiscoveryNode... nodes) {
Expand Down Expand Up @@ -621,11 +647,7 @@ PlainActionFuture<ComputeResponse> sendRequests(
void searchShards(Set<String> concreteIndices, ActionListener<TargetShards> listener) {
runWithDelay(
() -> listener.onResponse(
new TargetShards(
shards.stream().collect(Collectors.toMap(TargetShard::shardId, Function.identity())),
shards.size(),
0
)
new TargetShards(shards.stream().collect(toMap(TargetShard::shardId, Function.identity())), shards.size(), 0)
)
);
}
Expand Down