Skip to content

Early detection of circuit breaker exception in the coordinating node #67431

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 1 commit into from
Jan 15, 2021
Merged

Early detection of circuit breaker exception in the coordinating node #67431

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
16 changes: 11 additions & 5 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 @@ -316,6 +316,17 @@ public void consume(QuerySearchResult result, Runnable next) {
emptyResults.add(new SearchShard(target.getClusterAlias(), target.getShardId()));
}
} else {
if (hasAggs) {
long aggsSize = ramBytesUsedQueryResult(result);
try {
addEstimateAndMaybeBreak(aggsSize);
} catch (Exception exc) {
onMergeFailure(exc);
next.run();
return;
}
aggsCurrentBufferSize += aggsSize;
}
// add one if a partial merge is pending
int size = buffer.size() + (hasPartialReduce ? 1 : 0);
if (size >= batchReduceSize) {
Expand All @@ -329,11 +340,6 @@ public void consume(QuerySearchResult result, Runnable next) {
queue.add(task);
tryExecuteNext();
}
if (hasAggs) {
long aggsSize = ramBytesUsedQueryResult(result);
addWithoutBreaking(aggsSize);
aggsCurrentBufferSize += aggsSize;
}
buffer.add(result);
}
}
Expand Down
2 changes: 2 additions & 0 deletions server/src/main/java/org/elasticsearch/action/search/SearchQueryThenFetchAsyncAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -93,6 +93,8 @@ protected void onShardResult(SearchPhaseResult result, SearchShardIterator shard
if (queryResult.isNull() == false
// disable sort optims for scroll requests because they keep track of the last bottom doc locally (per shard)
&& getRequest().scroll() == null
// top docs are already consumed if the query was cancelled or in error.
&& queryResult.hasConsumedTopDocs() == false
&& queryResult.topDocs() != null
&& queryResult.topDocs().topDocs.getClass() == TopFieldDocs.class) {
TopFieldDocs topDocs = (TopFieldDocs) queryResult.topDocs().topDocs;
Expand Down
33 changes: 14 additions & 19 deletions server/src/test/java/org/elasticsearch/action/search/SearchPhaseControllerTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -936,23 +936,16 @@ public void onFinalReduce(List<SearchShard> shards, TotalHits totalHits, Interna
}
}

public void testPartialReduce() throws Exception {
for (int i = 0; i < 10; i++) {
testReduceCase(false);
}
}

public void testPartialReduceWithFailure() throws Exception {
for (int i = 0; i < 10; i++) {
testReduceCase(true);
}
public void testCoordCircuitBreaker() throws Exception {
int numShards = randomIntBetween(20, 200);
testReduceCase(numShards, numShards, true);
testReduceCase(numShards, numShards, false);
testReduceCase(numShards, randomIntBetween(2, numShards-1), true);
testReduceCase(numShards, randomIntBetween(2, numShards-1), false);
}

private void testReduceCase(boolean shouldFail) throws Exception {
int expectedNumResults = randomIntBetween(20, 200);
int bufferSize = randomIntBetween(2, expectedNumResults - 1);
private void testReduceCase(int numShards, int bufferSize, boolean shouldFail) throws Exception {
SearchRequest request = new SearchRequest();

request.source(new SearchSourceBuilder().aggregation(AggregationBuilders.avg("foo")).size(0));
request.setBatchedReduceSize(bufferSize);
AtomicBoolean hasConsumedFailure = new AtomicBoolean();
Expand All @@ -963,10 +956,10 @@ private void testReduceCase(boolean shouldFail) throws Exception {
}
QueryPhaseResultConsumer consumer = searchPhaseController.newSearchPhaseResults(fixedExecutor,
circuitBreaker, SearchProgressListener.NOOP,
request, expectedNumResults, exc -> hasConsumedFailure.set(true));
CountDownLatch latch = new CountDownLatch(expectedNumResults);
Thread[] threads = new Thread[expectedNumResults];
for (int i = 0; i < expectedNumResults; i++) {
request, numShards, exc -> hasConsumedFailure.set(true));
CountDownLatch latch = new CountDownLatch(numShards);
Thread[] threads = new Thread[numShards];
for (int i = 0; i < numShards; i++) {
final int index = i;
threads[index] = new Thread(() -> {
QuerySearchResult result = new QuerySearchResult(new ShardSearchContextId(UUIDs.randomBase64UUID(), index),
Expand All @@ -985,13 +978,15 @@ private void testReduceCase(boolean shouldFail) throws Exception {
});
threads[index].start();
}
for (int i = 0; i < expectedNumResults; i++) {
for (int i = 0; i < numShards; i++) {
threads[i].join();
}
latch.await();
if (shouldFail) {
if (shouldFailPartial == false) {
circuitBreaker.shouldBreak.set(true);
} else {
circuitBreaker.shouldBreak.set(false);
}
CircuitBreakingException exc = expectThrows(CircuitBreakingException.class, () -> consumer.reduce());
assertEquals(shouldFailPartial, hasConsumedFailure.get());
Expand Down