forked from opensearch-project/OpenSearch
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Make MultiBucketConsumerService thread safe to use across slices duri…
…ng search (opensearch-project#9047) Signed-off-by: Neetika Singhal <neetiks@amazon.com> (cherry picked from commit 24595c9)
- Loading branch information
1 parent
b3c8a94
commit 2bd707a
Showing
3 changed files
with
124 additions
and
6 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
72 changes: 72 additions & 0 deletions
72
server/src/test/java/org/opensearch/search/aggregations/MultiBucketConsumerTests.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,72 @@ | ||
| /* | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| * | ||
| * The OpenSearch Contributors require contributions made to | ||
| * this file be licensed under the Apache-2.0 license or a | ||
| * compatible open source license. | ||
| */ | ||
|
|
||
| package org.opensearch.search.aggregations; | ||
|
|
||
| import org.mockito.Mockito; | ||
| import org.opensearch.core.common.breaker.CircuitBreaker; | ||
| import org.opensearch.core.common.breaker.CircuitBreakingException; | ||
| import org.opensearch.test.OpenSearchTestCase; | ||
|
|
||
| import java.util.concurrent.atomic.LongAdder; | ||
|
|
||
| import static org.opensearch.search.aggregations.MultiBucketConsumerService.DEFAULT_MAX_BUCKETS; | ||
|
|
||
| public class MultiBucketConsumerTests extends OpenSearchTestCase { | ||
|
|
||
| public void testMultiConsumerAcceptWhenCBTripped() { | ||
| CircuitBreaker breaker = Mockito.mock(CircuitBreaker.class); | ||
| MultiBucketConsumerService.MultiBucketConsumer multiBucketConsumer = new MultiBucketConsumerService.MultiBucketConsumer( | ||
| DEFAULT_MAX_BUCKETS, | ||
| breaker, | ||
| new LongAdder(), | ||
| true, | ||
| 1 | ||
| ); | ||
| // exception is thrown upfront since the circuit breaker has already tripped | ||
| expectThrows(CircuitBreakingException.class, () -> multiBucketConsumer.accept(0)); | ||
| Mockito.verify(breaker, Mockito.times(0)).addEstimateBytesAndMaybeBreak(0, "allocated_buckets"); | ||
| } | ||
|
|
||
| public void testMultiConsumerAcceptToTripCB() { | ||
| CircuitBreaker breaker = Mockito.mock(CircuitBreaker.class); | ||
| LongAdder callCount = new LongAdder(); | ||
| for (int i = 0; i < 1024; i++) { | ||
| callCount.increment(); | ||
| } | ||
| MultiBucketConsumerService.MultiBucketConsumer multiBucketConsumer = new MultiBucketConsumerService.MultiBucketConsumer( | ||
| DEFAULT_MAX_BUCKETS, | ||
| breaker, | ||
| callCount, | ||
| false, | ||
| 2 | ||
| ); | ||
| // circuit breaker check is performed as the value of call count would be 1025 which is still in range | ||
| Mockito.when(breaker.addEstimateBytesAndMaybeBreak(0, "allocated_buckets")).thenThrow(CircuitBreakingException.class); | ||
| expectThrows(CircuitBreakingException.class, () -> multiBucketConsumer.accept(0)); | ||
| Mockito.verify(breaker, Mockito.times(1)).addEstimateBytesAndMaybeBreak(0, "allocated_buckets"); | ||
| } | ||
|
|
||
| public void testMultiConsumerAccept() { | ||
| CircuitBreaker breaker = Mockito.mock(CircuitBreaker.class); | ||
| LongAdder callCount = new LongAdder(); | ||
| for (int i = 0; i < 1100; i++) { | ||
| callCount.increment(); | ||
| } | ||
| MultiBucketConsumerService.MultiBucketConsumer multiBucketConsumer = new MultiBucketConsumerService.MultiBucketConsumer( | ||
| DEFAULT_MAX_BUCKETS, | ||
| breaker, | ||
| callCount, | ||
| false, | ||
| 1 | ||
| ); | ||
| // no exception is thrown as the call count value is not in the expected range and CB is not checked | ||
| multiBucketConsumer.accept(0); | ||
| Mockito.verify(breaker, Mockito.times(0)).addEstimateBytesAndMaybeBreak(0, "allocated_buckets"); | ||
| } | ||
| } |