Added SQS interactions for S3 source #1431

asifsmohammed · 2022-05-26T19:22:51Z

Description

Added sqs interaction to getMessage from SQS
And convert each Message to S3EventNotificationRecord
Added exponential backoff for reading messages from SQS

Used AWS SDK v1 for Deserializing SQS message into S3EventNotificationRecord since it's not yet ported to v2. For more info see: aws/aws-sdk-java-v2#1197

Issues Resolved

Resolves #1425

Check List

New functionality includes testing.
New functionality has been documented.
- New functionality has javadoc added
Commits are signed with a real name per the DCO

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

dlvenable · 2022-05-26T21:01:06Z

data-prepper-plugins/s3-source/build.gradle

@@ -13,10 +13,13 @@ repositories {

 dependencies {
    implementation project(':data-prepper-api')
-    implementation 'com.fasterxml.jackson.core:jackson-databind'
+    implementation project(':data-prepper-plugins:blocking-buffer')


Why did you add blocking-buffer here? Data Prepper API provides the Buffer interface.

I added this for testing S3Source using buffer.

For testing could this import be added as testImplementation project(':data-prepper-plugins:blocking-buffer')?

@sbayer55 's suggestion is the right approach for a testing dependency.

dlvenable · 2022-05-26T21:04:04Z

...prepper-plugins/s3-source/src/main/java/com/amazon/dataprepper/plugins/source/SqsWorker.java

+                messages.addAll(sqsClient.receiveMessage(receiveMessageRequest).messages());
+            } catch (SqsException e) {
+                LOG.error(e.awsErrorDetails().errorMessage());
+                System.exit(1);


This will exit all of Data Prepper. That is a far too severe consequence of an error communicating with SQS. I think that we want to log the exception, then wait some time, then try again. It could be appropriate to perform an exponential backoff.

dlvenable · 2022-05-26T21:05:07Z

...prepper-plugins/s3-source/src/main/java/com/amazon/dataprepper/plugins/source/SqsWorker.java

+        } while (true);
+    }
+
+    private S3EventNotification.S3EventNotificationRecord getS3EventMessages(Message message) {


I recommend renaming this so that it has a different verb than "get." Perhaps "convert" or "transform?" Get makes it seem that you are using a getter.

dlvenable · 2022-05-26T21:06:50Z

.../s3-source/src/main/java/com/amazon/dataprepper/plugins/source/configuration/SqsOptions.java

+public class SqsOptions {
+    private final int DEFAULT_MAXIMUM_MESSAGES = 10;
+    private final int DEFAULT_VISIBILITY_TIMEOUT_SECONDS = 30;
+    private final int DEFAULT_WAIT_TIME_SECONDS = 0;


I think this default should be 20 seconds (the current maximum).

dlvenable · 2022-05-26T21:07:32Z

.../s3-source/src/main/java/com/amazon/dataprepper/plugins/source/configuration/SqsOptions.java

+import java.time.Duration;
+
+public class SqsOptions {
+    private final int DEFAULT_MAXIMUM_MESSAGES = 10;


These should be private static final (They are missing the static).

dlvenable · 2022-05-26T21:09:50Z

.../s3-source/src/main/java/com/amazon/dataprepper/plugins/source/configuration/SqsOptions.java

+
+public class SqsOptions {
+    private final int DEFAULT_MAXIMUM_MESSAGES = 10;
+    private final int DEFAULT_VISIBILITY_TIMEOUT_SECONDS = 30;


Your choice here: You could make these defaults actual Duration objects. Since they will be static, you won't need to re-create the Duration in each field. I also think it is cleaner since the code only deals with Duration.

e.g.

private static final Duration DEFAULT_VISIBILITY_TIMEOUT = Duration.ofSeconds(20);

I like this suggestion.

dlvenable · 2022-05-26T21:11:22Z

...-prepper-plugins/s3-source/src/main/java/com/amazon/dataprepper/plugins/source/S3Source.java

-    private final S3Client s3Client;
-
-    private AwsCredentialsProvider createCredentialsProvider() {
+    public AwsCredentialsProvider createCredentialsProvider() {


Why is this public and not private?

dlvenable · 2022-05-26T21:12:06Z

...-prepper-plugins/s3-source/src/main/java/com/amazon/dataprepper/plugins/source/S3Source.java


+        for(int i = 0; i < s3SourceConfig.getSqsOptions().getThreadCount(); i++) {
+            Thread sqsWorkerThread = new Thread(new SqsWorker(sqsClient, s3Client, s3SourceConfig));


I think we will want to have a List<Thread> as a field. You can keep all the threads there. You will need to stop them when the stop() method for S3Source is called.

dlvenable · 2022-05-26T21:14:53Z

...prepper-plugins/s3-source/src/main/java/com/amazon/dataprepper/plugins/source/SqsWorker.java

+
+                messages.addAll(sqsClient.receiveMessage(receiveMessageRequest).messages());
+            } catch (SqsException e) {
+                LOG.error(e.awsErrorDetails().errorMessage());


In general you should avoid logging just an error message from some external source. It is better to have:

LOG.error("Error reading from SQS: {}", e.awsErrorDetails().errorMessage());

Relatedly - is awsErrorDetails() guaranteed to be non-null?

I'm not sure if awsErrorDetails() will be non-null. I'll check this one

dlvenable · 2022-05-26T21:16:20Z

...prepper-plugins/s3-source/src/main/java/com/amazon/dataprepper/plugins/source/SqsWorker.java

+                try {
+                    Thread.sleep(s3SourceConfig.getSqsOptions().getPollDelay().toMillis());
+                } catch (InterruptedException e) {
+                    e.printStackTrace();


We should log this, not call e.printStackTrace().

sbayer55 · 2022-05-31T16:53:47Z

As a general comment S3Source and SqsWorker have low test coverage. Are you planning to add functionality & test coverage to these classes in a future PR?

sbayer55 · 2022-05-31T16:16:33Z

data-prepper-plugins/s3-source/build.gradle

@@ -13,10 +13,13 @@ repositories {

 dependencies {
    implementation project(':data-prepper-api')
-    implementation 'com.fasterxml.jackson.core:jackson-databind'
+    implementation project(':data-prepper-plugins:blocking-buffer')


For testing could this import be added as testImplementation project(':data-prepper-plugins:blocking-buffer')?

sbayer55 · 2022-05-31T16:20:22Z

...pper-plugins/s3-source/src/main/java/com/amazon/dataprepper/plugins/source/BackoffUtils.java

+
+import java.util.Random;
+
+public class BackoffUtils {


In place of utils could BackoffUtils class have a more descriptive name? Consider something like RetryState or BackoffManager. Without looking at the class file I expected BackoffUtils to contain some helper functions for parsing or transforming data.

sbayer55 · 2022-05-31T16:23:41Z

...pper-plugins/s3-source/src/main/java/com/amazon/dataprepper/plugins/source/BackoffUtils.java

+
+        if (shouldRetry()) {
+            waitUntilNextTry();
+            timeToWait += random.nextInt(1000);


It there a benefit to a random wait opposed to a fixed time or exponential growth time?

Yeah, it's to break the synchronization across the clients thereby avoiding collisions

sbayer55 · 2022-05-31T16:26:35Z

...ns/s3-source/src/main/java/com/amazon/dataprepper/plugins/source/S3ClientAuthentication.java

+
+    public S3Client createS3Client(final StsClient stsClient) {
+
+        return software.amazon.awssdk.services.s3.S3Client.builder()


Can the fully qualified name be shortened from software.amazon.awssdk.services.s3.S3Client to S3Client? It looks like you already have the import.

cmanning09 · 2022-05-31T16:41:59Z

...pper-plugins/s3-source/src/main/java/com/amazon/dataprepper/plugins/source/BackoffUtils.java

+
+import java.util.Random;
+
+public class BackoffUtils {


Why are we opting to build our own exponential back off strategy? Is there a reason we cannot use the built-in AWS SDK retry functionality?

We do need a different mechanism in place for the entire thread. As the SQS processing runs in its own thread, it must not end until the the source is shutdown.

I think we should use both:

AWS SDK retry to help with transient errors

Catch exceptions in the thread, suppress, and wait. This goes beyond AWS SDK errors. Say, some message in the SQS queue is not actually JSON. We don't want the thread to fail for a single bad message.

That being said, perhaps this will not need any sophisticated back-off. Perhaps a straightforward Thread.sleep() is enough.

cmanning09 · 2022-05-31T17:15:18Z

...prepper-plugins/s3-source/src/main/java/com/amazon/dataprepper/plugins/source/SqsWorker.java

+    }
+
+    @Override
+    public void run() {


This function is getting very large in this partial state. I would encourage you to break it apart to improve maintainability.

This will also make it easier to unit test and can remove variable reassignment/mutation in the code below.

dlvenable · 2022-05-31T23:16:08Z

data-prepper-plugins/s3-source/build.gradle

@@ -13,10 +13,13 @@ repositories {

 dependencies {
    implementation project(':data-prepper-api')
-    implementation 'com.fasterxml.jackson.core:jackson-databind'
+    implementation project(':data-prepper-plugins:blocking-buffer')


@sbayer55 's suggestion is the right approach for a testing dependency.

dlvenable · 2022-05-31T23:18:24Z

...pper-plugins/s3-source/src/main/java/com/amazon/dataprepper/plugins/source/BackoffUtils.java

+
+import java.util.Random;
+
+public class BackoffUtils {


We do need a different mechanism in place for the entire thread. As the SQS processing runs in its own thread, it must not end until the the source is shutdown.

I think we should use both:

AWS SDK retry to help with transient errors

Catch exceptions in the thread, suppress, and wait. This goes beyond AWS SDK errors. Say, some message in the SQS queue is not actually JSON. We don't want the thread to fail for a single bad message.

That being said, perhaps this will not need any sophisticated back-off. Perhaps a straightforward Thread.sleep() is enough.

dlvenable · 2022-05-31T23:21:13Z

...-prepper-plugins/s3-source/src/main/java/com/amazon/dataprepper/plugins/source/S3Source.java

-                .region(Region.of(s3SourceConfig.getAWSAuthentication().getAwsRegion()))
-                .credentialsProvider(awsCredentialsProvider)
-                .build();
+        Thread.currentThread().interrupt();


Data Prepper core is calling the stop() method. Thus, this current code is going to interrupt the Data Prepper core thread.

Also, rather than use interrupt(), I'd like to have a running/shutdown boolean flag in the worker thread. You then can use while(running) { ... } instead of while(true) { ... }.

codecov-commenter · 2022-06-03T04:58:38Z

Codecov Report

Merging #1431 (3498a85) into main (a72618f) will not change coverage.
The diff coverage is n/a.

@@            Coverage Diff            @@
##               main    #1431   +/-   ##
=========================================
  Coverage     94.25%   94.25%           
  Complexity     1179     1179           
=========================================
  Files           165      165           
  Lines          3377     3377           
  Branches        276      276           
=========================================
  Hits           3183     3183           
  Misses          138      138           
  Partials         56       56

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update a72618f...3498a85. Read the comment docs.

dlvenable · 2022-06-03T15:50:38Z

...pper-plugins/s3-source/src/test/java/com/amazon/dataprepper/plugins/source/S3SourceTest.java

+        assertThrows(IllegalStateException.class, () -> s3Source.start(testBuffer));
+    }
+
+    private BlockingBuffer<Record<Event>> getBuffer() {


Is there a reason you can't use a mock here?

There may be value in creating some integration tests in other PRs which use the blocking buffer. But for a unit test, we should use mocks if at all possible.

dlvenable · 2022-06-03T15:55:13Z

...prepper-plugins/s3-source/src/main/java/com/amazon/dataprepper/plugins/source/SqsWorker.java

+                .build();
+    }
+
+    private S3EventNotification.S3EventNotificationRecord convertS3EventMessages(final Message message) {


I think we may want to make this part a separate class which is injected. One reason is that the SQS message may actually be in a different format in some cases. If the SQS queue is connected to SNS, then this becomes a string rather than JSON. So we might need to make this configurable by pipeline authors. I'm fine to keep it here for now if that helps merge it in though.

I think we can move this to a separate class once we decide on how pipeline authors can configure this.

dlvenable · 2022-06-03T16:00:37Z

...prepper-plugins/s3-source/src/main/java/com/amazon/dataprepper/plugins/source/SqsWorker.java

+            // build s3ObjectPointer from S3EventNotificationRecord if event name starts with ObjectCreated
+            List<S3ObjectReference> addedObjects = new ArrayList<>();
+            for (Map.Entry<Message, S3EventNotification.S3EventNotificationRecord> entry: s3EventNotificationRecords.entrySet()) {
+                if (entry.getValue().getEventName().startsWith("ObjectCreated")) {


I'd like to create an S3EventFilter structure. There are couple motivations I have in this:

We are also going to add an optional user-configuration to support filtering by bucketName or accountId.

If we add SNS as an option in future iterations, this filter should be the same code.

Here is a possible design:

public interface S3EventFilter { Optional<S3EventNotification.S3EventNotificationRecord> filter(S3EventNotification.S3EventNotificationRecord notification) } class ObjectCreatedFilter implements S3EventFilter { Optional<S3EventNotification.S3EventNotificationRecord> filter(S3EventNotification.S3EventNotificationRecord entry) { if (entry.getValue().getEventName().startsWith("ObjectCreated")) return Optional.of(entry); else return Optional.empty(); } }

We can build more complicated filters for the bucketName and accountId in another PR.

Signed-off-by: Asif Sohail Mohammed <nsifmoh@amazon.com>

dlvenable · 2022-06-03T20:03:56Z

...plugins/s3-source/src/main/java/com/amazon/dataprepper/plugins/source/S3ObjectReference.java

+
+import java.util.Objects;
+
+public class S3ObjectReference {


You might get a merge conflict on this file since I also added this into main. Please use the one I provided as it has toString() and is unit tested.

Signed-off-by: Asif Sohail Mohammed <nsifmoh@amazon.com>

cmanning09 · 2022-06-06T18:59:02Z

...-prepper-plugins/s3-source/src/main/java/com/amazon/dataprepper/plugins/source/S3Source.java

    }

    @Override
    public void start(Buffer<Record<Event>> buffer) {
+        if (buffer == null) {


The buffer is never used the by the S3Service or the SqsService. Who is going to be responsible for putting the item in the buffer (S3Service or SqsService)?

The S3ObjectWorker handles this:

data-prepper/data-prepper-plugins/s3-source/src/main/java/com/amazon/dataprepper/plugins/source/S3ObjectWorker.java

Line 30 in be35437

private final Buffer<Record<Event>> buffer;

Also, @asifsmohammed and I are coordinating on the work here. He created this PR before the S3ObjectWorker was present in main, so some portions like this are currently unused in his branch and work.

I'll update the logic with buffer here

cmanning09 · 2022-06-06T19:05:49Z

...per-plugins/s3-source/src/test/java/com/amazon/dataprepper/plugins/source/SqsWorkerTest.java

+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class SqsWorkerTest {


The run method is not tested. We are missing on out some core test coverage should add test cases.

cmanning09 · 2022-06-06T19:08:02Z

...prepper-plugins/s3-source/src/main/java/com/amazon/dataprepper/plugins/source/SqsWorker.java

+    }
+
+    @Override
+    public void run() {


This will also make it easier to unit test and can remove variable reassignment/mutation in the code below.

cmanning09 · 2022-06-06T19:08:50Z

...prepper-plugins/s3-source/src/main/java/com/amazon/dataprepper/plugins/source/SqsWorker.java

+    private static final Logger LOG = LoggerFactory.getLogger(SqsWorker.class);
+    private static final ObjectMapper objectMapper = new ObjectMapper();
+
+    private S3SourceConfig s3SourceConfig;


These private variables should be final to prevent re-assignment

cmanning09 · 2022-06-06T19:21:51Z

...prepper-plugins/s3-source/src/main/java/com/amazon/dataprepper/plugins/source/S3Service.java

+    S3ObjectReference addS3Object(final S3ObjectReference s3ObjectReference) {
+        // TODO: should return message id and receipt handle if successfully converted to event
+        return null;


I have a few concerns over this method and subsequent workflow. I have some ideas for improvements but I want to understand a few things first.

What are we adding an S3Object to? I find this method name confusing and was really confused by the usage in the run() method.

Why are the parameters and return object the same object (S3ObjectReference)? Are we extending S3ObjectReference and expecting this class to mutate the variable with a messageId and Receipt? (I want to see if we can avoid mutation and nullable variables)

How is the receipt used by the calling function? Do we need it?

Why are the parameters and return object the same object (S3ObjectReference)? Are we extending S3ObjectReference and expecting this class to mutate the variable with a messageId and Receipt? (I want to see if we can avoid mutation and nullable variables)

@asifsmohammed and I discussed a couple approaches here:

We can make this templated and add something like T getNotificationMetadata() which in this case returns SQS metadata. It could be extended for SNS metadata if that were ever added.

We could extend this class with a sub-class.

In either case, the SQS metadata would include the ReceiptHandle and MessageId.

I'm fine with either approach. But, the inheritance approach would have the benefit of completely abstracting this metadata from the S3 code.

My initial thought was to add ReceiptHandle and MessageId to S3ObjectReference which will be used to get S3 Object.
If S3 object is successfully processed we return the same S3ObjectReference to delete message from queue.

I'll create a subclass which contains ReceiptHandle and MessageId.

Signed-off-by: Asif Sohail Mohammed <nsifmoh@amazon.com>

cmanning09

Thanks for making those changes. Let's address the remaining in the next PR.

sbayer55

Thanks for the consistent updates @asifsmohammed !

* Added sqs configuration and basic sqs interactions Signed-off-by: Asif Sohail Mohammed <nsifmoh@amazon.com> Signed-off-by: Finn Roblin <finnrobl@amazon.com>

asifsmohammed requested a review from a team as a code owner May 26, 2022 19:22

asifsmohammed marked this pull request as draft May 26, 2022 19:22

dlvenable requested changes May 26, 2022

View reviewed changes

asifsmohammed marked this pull request as ready for review May 27, 2022 22:27

asifsmohammed force-pushed the sqs-interactions branch 2 times, most recently from 3816ace to f65d28f Compare May 31, 2022 06:50

sbayer55 suggested changes May 31, 2022

View reviewed changes

cmanning09 reviewed May 31, 2022

View reviewed changes

dlvenable reviewed May 31, 2022

View reviewed changes

asifsmohammed force-pushed the sqs-interactions branch from f65d28f to 234bcdb Compare June 2, 2022 19:08

dlvenable requested changes Jun 3, 2022

View reviewed changes

asifsmohammed added 9 commits June 3, 2022 14:53

Added sqs configuration

6c45969

Signed-off-by: Asif Sohail Mohammed <nsifmoh@amazon.com>

Added basic sqs interactions

cb834f8

Signed-off-by: Asif Sohail Mohammed <nsifmoh@amazon.com>

Added backoff and sqsclient auth

421010f

Signed-off-by: Asif Sohail Mohammed <nsifmoh@amazon.com>

Removed unused mocks from S3Source tests

ffcf989

Signed-off-by: Asif Sohail Mohammed <nsifmoh@amazon.com>

Added unit tests for sqsworker

fbc74f7

Signed-off-by: Asif Sohail Mohammed <nsifmoh@amazon.com>

Added separate classes for sqs and s3

391452a

Signed-off-by: Asif Sohail Mohammed <nsifmoh@amazon.com>

Added separate classes for sqs and s3

fdb2411

Signed-off-by: Asif Sohail Mohammed <nsifmoh@amazon.com>

Removed tests for creating s3 and sqs clients

2b77a9a

Signed-off-by: Asif Sohail Mohammed <nsifmoh@amazon.com>

Added interface for event filter

eb2bcca

Signed-off-by: Asif Sohail Mohammed <nsifmoh@amazon.com>

dlvenable reviewed Jun 3, 2022

View reviewed changes

dlvenable previously approved these changes Jun 3, 2022

View reviewed changes

Fixed merge conflicts

093a9a3

Signed-off-by: Asif Sohail Mohammed <nsifmoh@amazon.com>

asifsmohammed dismissed dlvenable’s stale review via 093a9a3 June 3, 2022 20:08

asifsmohammed force-pushed the sqs-interactions branch from 5a71690 to 093a9a3 Compare June 3, 2022 20:08

dlvenable previously approved these changes Jun 6, 2022

View reviewed changes

cmanning09 reviewed Jun 6, 2022

View reviewed changes

Added unit tests for SqsWorker

e387106

Signed-off-by: Asif Sohail Mohammed <nsifmoh@amazon.com>

asifsmohammed dismissed dlvenable’s stale review via e387106 June 7, 2022 15:59

cmanning09 approved these changes Jun 7, 2022

View reviewed changes

dlvenable approved these changes Jun 7, 2022

View reviewed changes

sbayer55 approved these changes Jun 7, 2022

View reviewed changes

asifsmohammed merged commit 8d1454c into opensearch-project:main Jun 7, 2022


		for(int i = 0; i < s3SourceConfig.getSqsOptions().getThreadCount(); i++) {
		Thread sqsWorkerThread = new Thread(new SqsWorker(sqsClient, s3Client, s3SourceConfig));


		public S3Client createS3Client(final StsClient stsClient) {

		return software.amazon.awssdk.services.s3.S3Client.builder()

Added SQS interactions for S3 source #1431

Added SQS interactions for S3 source #1431

Conversation

asifsmohammed commented May 26, 2022 • edited Loading

Description

Issues Resolved

Check List

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sbayer55 commented May 31, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codecov-commenter commented Jun 3, 2022 • edited Loading

Codecov Report

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cmanning09 left a comment

Choose a reason for hiding this comment

sbayer55 left a comment

Choose a reason for hiding this comment

asifsmohammed commented May 26, 2022 •

edited

Loading

codecov-commenter commented Jun 3, 2022 •

edited

Loading