Skip to content

Commit

Permalink
Modified wait times
Browse files Browse the repository at this point in the history 
Signed-off-by: Kondaka <[email protected]>
  • Loading branch information
@kkondaka
kkondaka committed Nov 27, 2023
1 parent 2c0b2a7 commit 1a35c10
Showing 1 changed file with 15 additions and 15 deletions.
30 changes: 15 additions & 15 deletions ...est/java/org/opensearch/dataprepper/plugins/processor/aggregate/AggregateProcessorIT.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -159,7 +159,7 @@ void aggregateWithNoConcludingGroupsReturnsExpectedResult() throws InterruptedEx
});
}

boolean allThreadsFinished = countDownLatch.await(5L, TimeUnit.SECONDS);
boolean allThreadsFinished = countDownLatch.await(10L, TimeUnit.SECONDS);

assertThat(allThreadsFinished, equalTo(true));
assertThat(aggregatedResult.size(), equalTo(NUM_UNIQUE_EVENTS_PER_BATCH));
Expand Down Expand Up @@ -198,7 +198,7 @@ void aggregateWithConcludingGroupsOnceReturnsExpectedResult() throws Interrupted
});
}

boolean allThreadsFinished = countDownLatch.await(5L, TimeUnit.SECONDS);
boolean allThreadsFinished = countDownLatch.await(10L, TimeUnit.SECONDS);

assertThat(allThreadsFinished, equalTo(true));
assertThat(aggregatedResult.size(), equalTo(NUM_UNIQUE_EVENTS_PER_BATCH));
Expand Down Expand Up @@ -236,7 +236,7 @@ void aggregateWithPutAllActionAndCondition() throws InterruptedException {
final CountDownLatch countDownLatch = new CountDownLatch(NUM_THREADS);

objectUnderTest.doExecute(eventBatch);
Thread.sleep(GROUP_DURATION_FOR_ONLY_SINGLE_CONCLUDE * 1000);
Thread.sleep(GROUP_DURATION_FOR_ONLY_SINGLE_CONCLUDE * 2000);

for (int i = 0; i < NUM_THREADS; i++) {
executorService.execute(() -> {
Expand All @@ -249,7 +249,7 @@ void aggregateWithPutAllActionAndCondition() throws InterruptedException {
});
}

boolean allThreadsFinished = countDownLatch.await(5L, TimeUnit.SECONDS);
boolean allThreadsFinished = countDownLatch.await(10L, TimeUnit.SECONDS);

assertThat(allThreadsFinished, equalTo(true));
assertThat(aggregatedResult.size(), equalTo(NUM_UNIQUE_EVENTS_PER_BATCH/2));
Expand Down Expand Up @@ -307,7 +307,7 @@ void aggregateWithRateLimiterAction() throws InterruptedException {
final CountDownLatch countDownLatch = new CountDownLatch(NUM_THREADS);

objectUnderTest.doExecute(eventBatch);
Thread.sleep(GROUP_DURATION_FOR_ONLY_SINGLE_CONCLUDE * 1000);
Thread.sleep(GROUP_DURATION_FOR_ONLY_SINGLE_CONCLUDE * 2000);

for (int i = 0; i < NUM_THREADS; i++) {
executorService.execute(() -> {
Expand All @@ -320,7 +320,7 @@ void aggregateWithRateLimiterAction() throws InterruptedException {
});
}

boolean allThreadsFinished = countDownLatch.await(5L, TimeUnit.SECONDS);
boolean allThreadsFinished = countDownLatch.await(10L, TimeUnit.SECONDS);

assertThat(allThreadsFinished, equalTo(true));
// Expect less number of events to be received, because of rate limiting
Expand All @@ -342,7 +342,7 @@ void aggregateWithRateLimiterActionNoDrops() throws InterruptedException {
final CountDownLatch countDownLatch = new CountDownLatch(NUM_THREADS);

objectUnderTest.doExecute(eventBatch);
Thread.sleep(GROUP_DURATION_FOR_ONLY_SINGLE_CONCLUDE * 1000);
Thread.sleep(GROUP_DURATION_FOR_ONLY_SINGLE_CONCLUDE * 2000);

for (int i = 0; i < NUM_THREADS; i++) {
executorService.execute(() -> {
Expand All @@ -355,7 +355,7 @@ void aggregateWithRateLimiterActionNoDrops() throws InterruptedException {
});
}

boolean allThreadsFinished = countDownLatch.await(5L, TimeUnit.SECONDS);
boolean allThreadsFinished = countDownLatch.await(10L, TimeUnit.SECONDS);

assertThat(allThreadsFinished, equalTo(true));
// Expect all events to be received even with rate limiting because no events are dropped
Expand Down Expand Up @@ -383,9 +383,9 @@ void aggregateWithCountAggregateAction() throws InterruptedException, NoSuchFiel
countDownLatch.countDown();
});
}
Thread.sleep(GROUP_DURATION_FOR_ONLY_SINGLE_CONCLUDE * 1000);
Thread.sleep(GROUP_DURATION_FOR_ONLY_SINGLE_CONCLUDE * 2000);

boolean allThreadsFinished = countDownLatch.await(5L, TimeUnit.SECONDS);
boolean allThreadsFinished = countDownLatch.await(10L, TimeUnit.SECONDS);
assertThat(allThreadsFinished, equalTo(true));

Collection<Record<Event>> results = objectUnderTest.doExecute(new ArrayList<Record<Event>>());
Expand Down Expand Up @@ -431,7 +431,7 @@ void aggregateWithCountAggregateActionWithCondition() throws InterruptedExceptio
}
Thread.sleep(GROUP_DURATION_FOR_ONLY_SINGLE_CONCLUDE * 2000);

boolean allThreadsFinished = countDownLatch.await(5L, TimeUnit.SECONDS);
boolean allThreadsFinished = countDownLatch.await(10L, TimeUnit.SECONDS);
assertThat(allThreadsFinished, equalTo(true));

Collection<Record<Event>> results = objectUnderTest.doExecute(new ArrayList<Record<Event>>());
Expand Down Expand Up @@ -491,7 +491,7 @@ void aggregateWithHistogramAggregateAction() throws InterruptedException, NoSuch
}
Thread.sleep(GROUP_DURATION_FOR_ONLY_SINGLE_CONCLUDE * 2000);

boolean allThreadsFinished = countDownLatch.await(5L, TimeUnit.SECONDS);
boolean allThreadsFinished = countDownLatch.await(10L, TimeUnit.SECONDS);
assertThat(allThreadsFinished, equalTo(true));

Collection<Record<Event>> results = objectUnderTest.doExecute(new ArrayList<Record<Event>>());
Expand Down Expand Up @@ -535,7 +535,7 @@ void aggregateWithTailSamplerAction(final int testPercent) throws InterruptedExc
final int numberOfSpans = 5;
eventBatch = getBatchOfEventsForTailSampling(numberOfErrorTraces, numberOfSpans);
objectUnderTest.doExecute(eventBatch);
Thread.sleep(GROUP_DURATION_FOR_ONLY_SINGLE_CONCLUDE * 1000);
Thread.sleep(GROUP_DURATION_FOR_ONLY_SINGLE_CONCLUDE * 2000);
final CountDownLatch countDownLatch = new CountDownLatch(NUM_THREADS);

for (int i = 0; i < NUM_THREADS; i++) {
Expand All @@ -544,8 +544,8 @@ void aggregateWithTailSamplerAction(final int testPercent) throws InterruptedExc
countDownLatch.countDown();
});
}
Thread.sleep(GROUP_DURATION_FOR_ONLY_SINGLE_CONCLUDE * 1000);
boolean allThreadsFinished = countDownLatch.await(5L, TimeUnit.SECONDS);
Thread.sleep(GROUP_DURATION_FOR_ONLY_SINGLE_CONCLUDE * 2000);
boolean allThreadsFinished = countDownLatch.await(10L, TimeUnit.SECONDS);
assertThat(allThreadsFinished, equalTo(true));
List<Event> errorEventList = eventBatch.stream().map(Record::getData).filter(event -> {
Event ev = ((Event)event);
Expand Down

0 comments on commit 1a35c10

Please sign in to comment.