Increase timeout values in in-memory source PipelinesWithAcksIT to fi…

…x occasional test failures (#2606) Signed-off-by: Krishna Kondaka <krishkdk@amazon.com>
opensearch-project · May 1, 2023 · 94974df · 94974df
1 parent f05c4b2
commit 94974df
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 10 deletions.
diff --git a/.../src/integrationTest/java/org/opensearch/dataprepper/integration/PipelinesWithAcksIT.java b/.../src/integrationTest/java/org/opensearch/dataprepper/integration/PipelinesWithAcksIT.java
@@ -60,7 +60,7 @@ void simple_pipeline_with_single_record() {
         final int numRecords = 1;
         inMemorySourceAccessor.submit(IN_MEMORY_IDENTIFIER, numRecords);
 
-        await().atMost(1000, TimeUnit.MILLISECONDS)
+        await().atMost(2000, TimeUnit.MILLISECONDS)
                 .untilAsserted(() -> {
             List<Record<Event>> outputRecords = inMemorySinkAccessor.get(IN_MEMORY_IDENTIFIER);
             assertThat(outputRecords, not(empty()));
@@ -76,7 +76,7 @@ void simple_pipeline_with_multiple_records() {
         final int numRecords = 100;
         inMemorySourceAccessor.submit(IN_MEMORY_IDENTIFIER, numRecords);
 
-        await().atMost(1000, TimeUnit.MILLISECONDS)
+        await().atMost(2000, TimeUnit.MILLISECONDS)
                 .untilAsserted(() -> {
             List<Record<Event>> outputRecords = inMemorySinkAccessor.get(IN_MEMORY_IDENTIFIER);
             assertThat(outputRecords, not(empty()));
@@ -106,7 +106,7 @@ void three_pipelines_with_multiple_records() {
         final int numRecords = 100;
         inMemorySourceAccessor.submit(IN_MEMORY_IDENTIFIER, numRecords);
 
-        await().atMost(10000, TimeUnit.MILLISECONDS)
+        await().atMost(20000, TimeUnit.MILLISECONDS)
                 .untilAsserted(() -> {
             List<Record<Event>> outputRecords = inMemorySinkAccessor.get(IN_MEMORY_IDENTIFIER);
             assertThat(outputRecords, not(empty()));
@@ -121,7 +121,7 @@ void three_pipelines_with_route_and_multiple_records() {
         final int numRecords = 100;
         inMemorySourceAccessor.submitWithStatus(IN_MEMORY_IDENTIFIER, numRecords);
 
-        await().atMost(10000, TimeUnit.MILLISECONDS)
+        await().atMost(20000, TimeUnit.MILLISECONDS)
                 .untilAsserted(() -> {
             List<Record<Event>> outputRecords = inMemorySinkAccessor.get(IN_MEMORY_IDENTIFIER);
             assertThat(outputRecords, not(empty()));
@@ -136,7 +136,7 @@ void two_parallel_pipelines_multiple_records() {
         final int numRecords = 100;
         inMemorySourceAccessor.submit(IN_MEMORY_IDENTIFIER, numRecords);
 
-        await().atMost(10000, TimeUnit.MILLISECONDS)
+        await().atMost(20000, TimeUnit.MILLISECONDS)
                 .untilAsserted(() -> {
             List<Record<Event>> outputRecords = inMemorySinkAccessor.get(IN_MEMORY_IDENTIFIER);
             assertThat(outputRecords, not(empty()));
@@ -151,7 +151,7 @@ void three_pipelines_multi_sink_multiple_records() {
         final int numRecords = 100;
         inMemorySourceAccessor.submit(IN_MEMORY_IDENTIFIER, numRecords);
 
-        await().atMost(10000, TimeUnit.MILLISECONDS)
+        await().atMost(20000, TimeUnit.MILLISECONDS)
                 .untilAsserted(() -> {
             List<Record<Event>> outputRecords = inMemorySinkAccessor.get(IN_MEMORY_IDENTIFIER);
             assertThat(outputRecords, not(empty()));
@@ -166,7 +166,7 @@ void one_pipeline_three_sinks_multiple_records() {
         final int numRecords = 100;
         inMemorySourceAccessor.submit(IN_MEMORY_IDENTIFIER, numRecords);
 
-        await().atMost(10000, TimeUnit.MILLISECONDS)
+        await().atMost(20000, TimeUnit.MILLISECONDS)
                 .untilAsserted(() -> {
             List<Record<Event>> outputRecords = inMemorySinkAccessor.get(IN_MEMORY_IDENTIFIER);
             assertThat(outputRecords, not(empty()));
@@ -181,7 +181,7 @@ void one_pipeline_ack_expiry_multiple_records() {
         final int numRecords = 100;
         inMemorySourceAccessor.submit(IN_MEMORY_IDENTIFIER, numRecords);
 
-        await().atMost(10000, TimeUnit.MILLISECONDS)
+        await().atMost(20000, TimeUnit.MILLISECONDS)
           .untilAsserted(() -> {
             List<Record<Event>> outputRecords = inMemorySinkAccessor.get(IN_MEMORY_IDENTIFIER);
             assertThat(outputRecords, not(empty()));
@@ -197,7 +197,7 @@ void one_pipeline_three_sinks_negative_ack_multiple_records() {
         inMemorySourceAccessor.submit(IN_MEMORY_IDENTIFIER, numRecords);
         inMemorySinkAccessor.setResult(false);
 
-        await().atMost(10000, TimeUnit.MILLISECONDS)
+        await().atMost(20000, TimeUnit.MILLISECONDS)
                 .untilAsserted(() -> {
             List<Record<Event>> outputRecords = inMemorySinkAccessor.get(IN_MEMORY_IDENTIFIER);
             assertThat(outputRecords, not(empty()));

diff --git a/...pper-core/src/integrationTest/java/org/opensearch/dataprepper/plugins/InMemorySource.java b/...pper-core/src/integrationTest/java/org/opensearch/dataprepper/plugins/InMemorySource.java
@@ -119,7 +119,7 @@ public void run() {
                                 {
                                     inMemorySourceAccessor.setAckReceived(result);
                                 },
-                                Duration.ofSeconds(5));
+                                Duration.ofSeconds(15));
                     records.stream().forEach((record) -> { ackSet.add(record.getData()); });
                     writeToBuffer(records);
                 } catch (final Exception ex) {