Skip to content

Fix AsyncSearchActionIT tests #127010

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
Apr 17, 2025
Merged

Fix AsyncSearchActionIT tests #127010

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
080dcd7
Fix AsyncSearchActionIT tests
original-brownbear Apr 17, 2025
a3c54d2
Merge remote-tracking branch 'elastic/main' into fix-async-search-its
original-brownbear Apr 17, 2025
c675b04
Merge branch 'main' into fix-async-search-its
original-brownbear Apr 17, 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
21 changes: 0 additions & 21 deletions muted-tests.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -408,27 +408,6 @@ tests:
- class: org.elasticsearch.cli.keystore.AddStringKeyStoreCommandTests
method: testStdinWithMultipleValues
issue: https://github.com/elastic/elasticsearch/issues/126882
- class: org.elasticsearch.xpack.search.AsyncSearchActionIT
method: testDeleteCancelRunningTask
issue: https://github.com/elastic/elasticsearch/issues/126994
- class: org.elasticsearch.xpack.search.AsyncSearchActionIT
method: testMaxResponseSize
issue: https://github.com/elastic/elasticsearch/issues/126995
- class: org.elasticsearch.xpack.search.AsyncSearchActionIT
method: testRemoveAsyncIndex
issue: https://github.com/elastic/elasticsearch/issues/126975
- class: org.elasticsearch.xpack.search.AsyncSearchActionIT
method: testCleanupOnFailure
issue: https://github.com/elastic/elasticsearch/issues/126999
- class: org.elasticsearch.xpack.search.AsyncSearchActionIT
method: testUpdateStoreKeepAlive
issue: https://github.com/elastic/elasticsearch/issues/127001
- class: org.elasticsearch.xpack.search.AsyncSearchActionIT
method: testRestartAfterCompletion
issue: https://github.com/elastic/elasticsearch/issues/126974
- class: org.elasticsearch.xpack.search.AsyncSearchActionIT
method: testDeleteCleanupIndex
issue: https://github.com/elastic/elasticsearch/issues/127008
- class: org.elasticsearch.packaging.test.DockerTests
method: test024InstallPluginFromArchiveUsingConfigFile
issue: https://github.com/elastic/elasticsearch/issues/126936
Expand Down
15 changes: 11 additions & 4 deletions server/src/main/java/org/elasticsearch/action/search/QueryPhaseResultConsumer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
import org.elasticsearch.common.util.concurrent.AbstractRunnable;
import org.elasticsearch.core.Nullable;
import org.elasticsearch.core.Releasable;
import org.elasticsearch.core.Releasables;
import org.elasticsearch.core.Tuple;
import org.elasticsearch.search.SearchPhaseResult;
import org.elasticsearch.search.SearchService;
Expand Down Expand Up @@ -162,7 +163,7 @@ public void consumeResult(SearchPhaseResult result, Runnable next) {
consume(querySearchResult, next);
}

private final List<Tuple<TopDocsStats, MergeResult>> batchedResults = new ArrayList<>();
private final ArrayDeque<Tuple<TopDocsStats, MergeResult>> batchedResults = new ArrayDeque<>();

/**
* Unlinks partial merge results from this instance and returns them as a partial merge result to be sent to the coordinating node.
Expand Down Expand Up @@ -214,7 +215,7 @@ public SearchPhaseController.ReducedQueryPhase reduce() throws Exception {
buffer.sort(RESULT_COMPARATOR);
final TopDocsStats topDocsStats = this.topDocsStats;
var mergeResult = this.mergeResult;
final List<Tuple<TopDocsStats, MergeResult>> batchedResults;
final ArrayDeque<Tuple<TopDocsStats, MergeResult>> batchedResults;
synchronized (this.batchedResults) {
batchedResults = this.batchedResults;
}
Expand All @@ -226,8 +227,8 @@ public SearchPhaseController.ReducedQueryPhase reduce() throws Exception {
if (mergeResult != null) {
consumePartialMergeResult(mergeResult, topDocsList, aggsList);
}
for (int i = 0; i < batchedResults.size(); i++) {
Tuple<TopDocsStats, MergeResult> batchedResult = batchedResults.set(i, null);
Tuple<TopDocsStats, MergeResult> batchedResult;
while ((batchedResult = batchedResults.poll()) != null) {
topDocsStats.add(batchedResult.v1());
consumePartialMergeResult(batchedResult.v2(), topDocsList, aggsList);
}
Expand Down Expand Up @@ -528,6 +529,12 @@ private void releaseBuffer() {
querySearchResult.releaseAggs();
}
}
synchronized (this.batchedResults) {
Tuple<TopDocsStats, MergeResult> batchedResult;
while ((batchedResult = batchedResults.poll()) != null) {
Releasables.close(batchedResult.v2().reducedAggs());
}
}
}

private synchronized void onMergeFailure(Exception exc) {
Expand Down