Fixes a bug in the SqsWorkerIT where an NPE occurs. Updated the READM…

…E.md to include the new steps. (#1538) (#1540) Signed-off-by: David Venable <[email protected]> (cherry picked from commit bd00a72) Co-authored-by: David Venable <[email protected]>
opensearch-project · Jun 24, 2022 · 20dd0dd · 20dd0dd
1 parent c4e9e85
commit 20dd0dd
Show file tree

Hide file tree

Showing 2 changed files with 16 additions and 8 deletions.
diff --git a/data-prepper-plugins/s3-source/README.md b/data-prepper-plugins/s3-source/README.md
@@ -101,5 +101,5 @@ The integration tests for this plugin do not run as part of the Data Prepper bui
 The following command runs the integration tests:
 
 ```
-./gradlew :data-prepper-plugins:s3-source:integrationTest -Dtests.s3source.region=<your-aws-region> -Dtests.s3source.bucket=<your-bucket>
+./gradlew :data-prepper-plugins:s3-source:integrationTest -Dtests.s3source.region=<your-aws-region> -Dtests.s3source.bucket=<your-bucket> -Dtests.s3source.queue.url=<your-queue-url>
 ```
diff --git a/...s3-source/src/integrationTest/java/com/amazon/dataprepper/plugins/source/SqsWorkerIT.java b/...s3-source/src/integrationTest/java/com/amazon/dataprepper/plugins/source/SqsWorkerIT.java
@@ -8,6 +8,8 @@
 import com.amazon.dataprepper.metrics.PluginMetrics;
 import com.amazon.dataprepper.plugins.source.configuration.OnErrorOption;
 import com.amazon.dataprepper.plugins.source.configuration.SqsOptions;
+import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.Timer;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -28,12 +30,13 @@
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 import static org.hamcrest.Matchers.lessThanOrEqualTo;
 import static org.hamcrest.core.StringStartsWith.startsWith;
+import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.atLeastOnce;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
-public class SqsWorkerIT {
+class SqsWorkerIT {
     private SqsClient sqsClient;
     private S3Service s3Service;
     private S3SourceConfig s3SourceConfig;
@@ -57,6 +60,11 @@ void setUp() {
         s3Service = mock(S3Service.class);
 
         pluginMetrics = mock(PluginMetrics.class);
+        final Counter sharedCounter = mock(Counter.class);
+        final Timer sqsMessageDelayTimer = mock(Timer.class);
+
+        when(pluginMetrics.counter(anyString())).thenReturn(sharedCounter);
+        when(pluginMetrics.timer(anyString())).thenReturn(sqsMessageDelayTimer);
 
         final SqsOptions sqsOptions = mock(SqsOptions.class);
         when(sqsOptions.getSqsUrl()).thenReturn(System.getProperty("tests.s3source.queue.url"));
@@ -81,12 +89,12 @@ void processRemainingMessages() {
         while (sqsMessagesProcessed > 0);
     }
 
-    /*
-    * receiveMessage of SQS doesn't return the exact number of objects that are written to S3 even if long polling is enabled with
-    * MaxNumberOfMessages greater than the number of objects written.
-    * The default behaviour is it returns the message immediately as soon as there's a single message and can return upto MaxNumberOfMessages.
-    * So the asserts in this test verify at least one message is returned.
-    */
+    /**
+     * receiveMessage of SQS doesn't return the exact number of objects that are written to S3 even if long polling is enabled with
+     * MaxNumberOfMessages greater than the number of objects written.
+     * The default behaviour is it returns the message immediately as soon as there's a single message and can return upto MaxNumberOfMessages.
+     * So the asserts in this test verify at least one message is returned.
+     */
     @ParameterizedTest
     @ValueSource(ints = {1, 5})
     void processSqsMessages_should_return_at_least_one_message(final int numberOfObjectsToWrite) throws IOException {