Skip to content

Fix request processing scheduling #127464

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 10 commits into from
May 7, 2025
Merged

Fix request processing scheduling #127464

Show file tree
Hide file tree
Changes from 6 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
de2f695
Fix request processing scheduling
idegtiarenko Apr 28, 2025
a5fde53
Merge branch 'main' into debug_DataNodeRequestSenderIT
idegtiarenko Apr 28, 2025
7780ed7
fix retry limit
idegtiarenko Apr 28, 2025
3425c26
Merge branch 'main' into debug_DataNodeRequestSenderIT
idegtiarenko Apr 28, 2025
b2dd3da
replace `isHeldByCurrentThread`check
idegtiarenko Apr 29, 2025
90d297c
Merge branch 'main' into debug_DataNodeRequestSenderIT
idegtiarenko Apr 29, 2025
733e5c8
Merge branch 'main' into debug_DataNodeRequestSenderIT
idegtiarenko May 6, 2025
8a0dcc6
Move retry scheduling to trySendingRequestsForPendingShards
idegtiarenko May 6, 2025
b193ec8
handle all shards unassigned
idegtiarenko May 6, 2025
95ed34f
Merge branch 'main' into debug_DataNodeRequestSenderIT
idegtiarenko May 6, 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
3 changes: 0 additions & 3 deletions muted-tests.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -414,9 +414,6 @@ tests:
- class: org.elasticsearch.packaging.test.DockerTests
method: test026InstallBundledRepositoryPluginsViaConfigFile
issue: https://github.com/elastic/elasticsearch/issues/127158
- class: org.elasticsearch.xpack.esql.plugin.DataNodeRequestSenderTests
method: testRetryOnlyMovedShards
issue: https://github.com/elastic/elasticsearch/issues/127168
- class: org.elasticsearch.xpack.esql.plugin.DataNodeRequestSenderIT
method: testSearchWhileRelocating
issue: https://github.com/elastic/elasticsearch/issues/127188
Expand Down
17 changes: 11 additions & 6 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 @@ -257,17 +257,16 @@ private void sendOneNodeRequest(TargetShards targetShards, ComputeListener compu
final ActionListener<DriverCompletionInfo> listener = computeListener.acquireCompute();
sendRequest(request.node, request.shardIds, request.aliasFilters, new NodeListener() {

private final Set<ShardId> pendingRetries = new HashSet<>();
private Set<ShardId> pendingRetries;

void onAfter(DriverCompletionInfo info) {
nodePermits.get(request.node).release();
if (concurrentRequests != null) {
concurrentRequests.release();
}

if (pendingRetries.isEmpty() == false && remainingUnavailableShardResolutionAttempts.decrementAndGet() >= 0) {
if (pendingRetries != null) {
try {
sendingLock.lock();
var resolutions = resolveShards(pendingRetries);
for (var entry : resolutions.entrySet()) {
targetShards.shards.get(entry.getKey()).remainingNodes.addAll(entry.getValue());
Expand All @@ -291,19 +290,19 @@ public void onResponse(DataNodeComputeResponse response) {
}
for (var entry : response.shardLevelFailures().entrySet()) {
final ShardId shardId = entry.getKey();
maybeScheduleRetry(shardId, false, entry.getValue());
trackShardLevelFailure(shardId, false, entry.getValue());
pendingShardIds.add(shardId);
maybeScheduleRetry(shardId, false, entry.getValue());
}
onAfter(response.completionInfo());
}

@Override
public void onFailure(Exception e, boolean receivedData) {
for (ShardId shardId : request.shardIds) {
maybeScheduleRetry(shardId, receivedData, e);
trackShardLevelFailure(shardId, receivedData, e);
pendingShardIds.add(shardId);
maybeScheduleRetry(shardId, receivedData, e);
}
onAfter(DriverCompletionInfo.EMPTY);
}
Expand All @@ -322,7 +321,13 @@ private void maybeScheduleRetry(ShardId shardId, boolean receivedData, Exception
if (receivedData == false
&& targetShards.getShard(shardId).remainingNodes.isEmpty()
&& unwrapFailure(shardId, e) instanceof NoShardAvailableActionException) {
pendingRetries.add(shardId);
if (pendingRetries == null && remainingUnavailableShardResolutionAttempts.decrementAndGet() >= 0) {
pendingRetries = new HashSet<>();
sendingLock.lock();
Copy link
Member

@dnhatn dnhatn Apr 29, 2025

Choose a reason for hiding this comment

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

I am concerned that we are scattering sendingLock#lock and sendingLock#unlock in two different places. Can we keep them close?

Copy link
Contributor Author

@idegtiarenko idegtiarenko Apr 30, 2025 •
edited
Loading

Choose a reason for hiding this comment

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

They are in the same inner listener, guarding the same structure at the moment.
They were previously in the same method before but that was not enough and caused a bug.
I suspect we could do this by creating a releasable inner wrapper class on top of pendingRetries = new HashSet<>() but that sounds like an overkill not really helping with readability.

Copy link
Member

@dnhatn dnhatn Apr 30, 2025

Choose a reason for hiding this comment

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

@idegtiarenko Sorry, I should have provided more detail. My concern is that we acquire the sending lock in maybeScheduleRetry and release it in onAfter, which is linked to the status of pendingRetries. While the implementation is technically correct, I think we should stick to the simplest lock pattern unless there is a strong reason to do otherwise:

lock/tryLock
try {
  ...
} finally {
  unlock
}

I think we can follow this lock pattern in the DataNodeRequestSender class.

Copy link
Contributor Author

@idegtiarenko idegtiarenko May 5, 2025

Choose a reason for hiding this comment

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

This sounds a lot like the original implementation (within onAfter) that was not correct.
Do you see a way how to implement this suggestion while still locking conditionally (only if shard movement is detected) and blocking concurrent requests while it is detected that new shard location resolution is required?

Copy link
Contributor Author

@idegtiarenko idegtiarenko May 6, 2025

Choose a reason for hiding this comment

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

As discussed yesterday, I moved retry scheduling to trySendingRequestsForPendingShards in 8a0dcc6

}
if (pendingRetries != null) {
pendingRetries.add(shardId);
}
}
}
});
Expand Down