feat: SQS Partial batch Utility

docs/content/utilities/batch.mdx

@@ -0,0 +1,251 @@
+---
+title: SQS Batch Processing
+description: Utility
+---
+
+import Note from "../../src/components/Note"
+
+The SQS batch processing utility provides a way to handle partial failures when processing batches of messages from SQS.
+
+**Key Features**
+
+* Prevent successfully processed messages from being returned to SQS
+* A simple interface for individually processing messages from a batch
+
+**Background**
+
+When using SQS as a Lambda event source mapping, Lambda functions can be triggered with a batch of messages from SQS.
+
+If your function fails to process any message from the batch, the entire batch returns to your SQS queue, and your Lambda function will be triggered with the same batch again.
+
+With this utility, messages within a batch will be handled individually - only messages that were not successfully processed
+are returned to the queue.
+
+<Note type="warning">
+  While this utility lowers the chance of processing messages more than once, it is not guaranteed. We recommend implementing processing logic in an idempotent manner wherever possible.
+  <br/><br/>
+  More details on how Lambda works with SQS can be found in the <a href="https://docs.aws.amazon.com/lambda/latest/dg/with-sqs.html">AWS documentation</a>
+</Note>
+
+## Install
+
+To install this utility, add the following dependency to your project.
+
+```xml
+<dependency>
+    <groupId>software.amazon.lambda</groupId>
+    <artifactId>powertools-sqs</artifactId>
+    <version>0.4.0-beta</version>
+</dependency>
+```
+
+And configure the aspectj-maven-plugin to compile-time weave (CTW) the
+aws-lambda-powertools-java aspects into your project. You may already have this
+plugin in your pom. In that case add the dependency to the `aspectLibraries`
+section.
+
+```xml
+<build>
+    <plugins>
+        ...
+        <plugin>
+             <groupId>org.codehaus.mojo</groupId>
+             <artifactId>aspectj-maven-plugin</artifactId>
+             <version>1.11</version>
+             <configuration>
+                 <source>1.8</source>
+                 <target>1.8</target>
+                 <complianceLevel>1.8</complianceLevel>
+                 <aspectLibraries>
+                     <!--    highlight-start    -->
+                     <aspectLibrary>
+                         <groupId>software.amazon.lambda</groupId>
+                         <artifactId>powertools-sqs</artifactId>
+                     </aspectLibrary>
+                     <!--    highlight-end    -->
+                 </aspectLibraries>
+             </configuration>
+             <executions>
+                 <execution>
+                     <goals>
+                         <goal>compile</goal>
+                     </goals>
+                 </execution>
+             </executions>
+        </plugin>
+        ...
+    </plugins>
+</build>
+```
+
+**IAM Permissions**
+
+This utility requires additional permissions to work as expected. Lambda functions using this utility require the `sqs:GetQueueUrl` and `sqs:DeleteMessageBatch` permission.
+
+## Processing messages from SQS
+
+You can use either **[SqsBatchProcessor annotation](#SqsBatchProcessor annotation)**, or **[PowertoolsSqs Utility API](#PowertoolsSqs Utility API)** as a fluent API.
+
+Both have nearly the same behaviour when it comes to processing messages from the batch:
+
+* **Entire batch has been successfully processed**, where your Lambda handler returned successfully, we will let SQS delete the batch to optimize your cost
+* **Entire Batch has been partially processed successfully**, where exceptions were raised within your `SqsMessageHandler` interface implementation, we will:
+    - **1)** Delete successfully processed messages from the queue by directly calling `sqs:DeleteMessageBatch`
+    - **2)** Raise `SQSBatchProcessingException` to ensure failed messages return to your SQS queue
+
+The only difference is that **PowertoolsSqs Utility API** will give you access to return from the processed messages if you need. Exception `SQSBatchProcessingException` thrown from the
+utility will have access to both successful and failed messaged along with failure exceptions.
+
+## Functional Interface SqsMessageHandler
+
+Both [annotation](#SqsBatchProcessor annotation) and [PowertoolsSqs Utility API](#PowertoolsSqs Utility API) requires an implementation of functional interface `SqsMessageHandler`.
+
+This implementation is responsible for processing each individual message from the batch, and to raise an exception if unable to process any of the messages sent.
+
+**Any non-exception/successful return from your record handler function** will instruct utility to queue up each individual message for deletion.
+
+### SqsBatchProcessor annotation
+
+When using this annotation, you need provide a class implementation of `SqsMessageHandler` that will process individual messages from the batch - It should raise an exception if it is unable to process the record.
+
+All records in the batch will be passed to this handler for processing, even if exceptions are thrown - Here's the behaviour after completing the batch:
+
+* **Any successfully processed messages**, we will delete them from the queue via `sqs:DeleteMessageBatch`
+* **Any unprocessed messages detected**, we will raise `SQSBatchProcessingException` to ensure failed messages return to your SQS queue
+
+<Note type="warning">
+  You will not have accessed to the <strong>processed messages</strong> within the Lambda Handler - all processing logic will and should be performed by the implemented <code>SqsMessageHandler#process()</code> function.
+
+</Note><br/>
+
+```java:title=App.java
+public class AppSqsEvent implements RequestHandler<SQSEvent, String> {
+    @Override
+    @SqsBatchProcessor(SampleMessageHandler.class) // highlight-line
+    public String handleRequest(SQSEvent input, Context context) {
+        return "{\"statusCode\": 200}";
+    }
+
+    public class SampleMessageHandler implements SqsMessageHandler<Object> {
+
+        @Override
+        public String process(SQSMessage message) {
+            // This will be called for each individual message from a batch
+            // It should raise an exception if the message was not processed successfully
+            String returnVal = doSomething(message.getBody());
+            return returnVal;
+        }
+    }
+}
+```
+
+### PowertoolsSqs Utility API
+
+If you require access to the result of processed messages, you can use this utility.
+
+The result from calling <code>PowertoolsSqs#batchProcessor()</code> on the context manager will be a list of all the return values from your <code>SqsMessageHandler#process()</code>  function.
+
+```java:title=App.java
+public class AppSqsEvent implements RequestHandler<SQSEvent, List<String>> {
+    @Override
+    public List<String> handleRequest(SQSEvent input, Context context) {
+        List<String> returnValues = PowertoolsSqs.batchProcessor(input, SampleMessageHandler.class); // highlight-line
+
+        return returnValues;
+    }
+
+    public class SampleMessageHandler implements SqsMessageHandler<String> {
+
+        @Override
+        public String process(SQSMessage message) {
+            // This will be called for each individual message from a batch
+            // It should raise an exception if the message was not processed successfully
+            String returnVal = doSomething(message.getBody());
+            return returnVal;
+        }
+    }
+}
+```
+
+You can also use the utility in a more functional way` by providing inline implementation of functional interface <code>SqsMessageHandler#process()</code>
+
+```java:title=App.java
+public class AppSqsEvent implements RequestHandler<SQSEvent, List<String>> {
+
+    @Override
+    public List<String> handleRequest(SQSEvent input, Context context) {
+        // highlight-start
+        List<String> returnValues = PowertoolsSqs.batchProcessor(input, (message) -> {
+            // This will be called for each individual message from a batch
+            // It should raise an exception if the message was not processed successfully
+            String returnVal = doSomething(message.getBody());
+            return returnVal;
+        });
+        // highlight-end
+
+        return returnValues;
+    }
+}
+```
+
+## Passing custom SqsClient
+
+If you need to pass custom SqsClient such as region to the SDK, you can pass your own `SqsClient` to be used by utility either for
+**[SqsBatchProcessor annotation](#SqsBatchProcessor annotation)**, or **[PowertoolsSqs Utility API](#PowertoolsSqs Utility API)**.
+
+```java:title=App.java
+
+public class AppSqsEvent implements RequestHandler<SQSEvent, List<String>> {
+    // highlight-start
+    static {
+        PowertoolsSqs.overrideSqsClient(SqsClient.builder()
+                .build());
+    }
+    // highlight-end
+
+    @Override
+    public List<String> handleRequest(SQSEvent input, Context context) {
+        List<String> returnValues = PowertoolsSqs.batchProcessor(input, SampleMessageHandler.class);
+
+        return returnValues;
+    }
+
+    public class SampleMessageHandler implements SqsMessageHandler<String> {
+
+        @Override
+        public String process(SQSMessage message) {
+            // This will be called for each individual message from a batch
+            // It should raise an exception if the message was not processed successfully
+            String returnVal = doSomething(message.getBody());
+            return returnVal;
+        }
+    }
+}
+
+```
+
+## Suppressing exceptions
+
+If you want to disable the default behavior where `SQSBatchProcessingException` is raised if there are any exception, you can pass the `suppressException` boolean argument.
+
+**Within SqsBatchProcessor annotation**
+
+```java:title=App.java
+...
+    @Override
+    @SqsBatchProcessor(value = SampleMessageHandler.class, suppressException = true) // highlight-line
+    public String handleRequest(SQSEvent input, Context context) {
+        return "{\"statusCode\": 200}";
+    }
+```
+
+**Within PowertoolsSqs Utility API**
+
+```java:title=App.java
+    @Override
+    public List<String> handleRequest(SQSEvent input, Context context) {
+        List<String> returnValues = PowertoolsSqs.batchProcessor(input, true, SampleMessageHandler.class); // highlight-line
+
+        return returnValues;
+    }
+```

docs/content/utilities/sqs_large_message_handling.mdx

@@ -35,7 +35,7 @@ To install this utility, add the following dependency to your project.
 
 And configure the aspectj-maven-plugin to compile-time weave (CTW) the
 aws-lambda-powertools-java aspects into your project. You may already have this
-plugin in your pom. In that case add the depenedency to the `aspectLibraries`
+plugin in your pom. In that case add the dependency to the `aspectLibraries`
 section.
 
 ```xml
@@ -51,12 +51,12 @@ section.
                  <target>1.8</target>
                  <complianceLevel>1.8</complianceLevel>
                  <aspectLibraries>
-                     ...
+                    <!--    highlight-start    -->
                      <aspectLibrary>
                          <groupId>software.amazon.lambda</groupId>
                          <artifactId>powertools-sqs</artifactId>
                      </aspectLibrary>
-                     ...
+                     <!--    highlight-end    -->
                  </aspectLibraries>
              </configuration>
              <executions>

docs/gatsby-config.js

@@ -29,7 +29,8 @@ module.exports = {
                     ],
                     'Utilities': [
                         'utilities/sqs_large_message_handling',
-                        'utilities/parameters'
+                        'utilities/batch',
+                        'utilities/parameters',
                     ],
                 },
                 navConfig: {

example/HelloWorldFunction/pom.xml

@@ -33,6 +33,11 @@
             <artifactId>powertools-parameters</artifactId>
             <version>0.4.0-beta</version>
         </dependency>
+        <dependency>
+            <groupId>software.amazon.lambda</groupId>
+            <artifactId>powertools-sqs</artifactId>
+            <version>0.4.0-beta</version>
+        </dependency>
         <dependency>
             <groupId>com.amazonaws</groupId>
             <artifactId>aws-lambda-java-core</artifactId>
@@ -90,6 +95,10 @@
                          <groupId>software.amazon.lambda</groupId>
                          <artifactId>powertools-metrics</artifactId>
                      </aspectLibrary>
+                     <aspectLibrary>
+                         <groupId>software.amazon.lambda</groupId>
+                         <artifactId>powertools-sqs</artifactId>
+                     </aspectLibrary>
                  </aspectLibraries>
              </configuration>
              <executions>

example/HelloWorldFunction/src/main/java/helloworld/AppSqsEvent.java

@@ -0,0 +1,35 @@
+package helloworld;
+
+import com.amazonaws.services.lambda.runtime.Context;
+import com.amazonaws.services.lambda.runtime.RequestHandler;
+import com.amazonaws.services.lambda.runtime.events.SQSEvent;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import software.amazon.lambda.powertools.logging.PowertoolsLogging;
+import software.amazon.lambda.powertools.sqs.SqsBatchProcessor;
+import software.amazon.lambda.powertools.sqs.SqsMessageHandler;
+
+import static com.amazonaws.services.lambda.runtime.events.SQSEvent.SQSMessage;
+
+public class AppSqsEvent implements RequestHandler<SQSEvent, String> {
+    private static final Logger LOG = LogManager.getLogger(AppSqsEvent.class);
+
+    @Override
+    @SqsBatchProcessor(SampleMessageHandler.class)
+    @PowertoolsLogging(logEvent = true)
+    public String handleRequest(SQSEvent input, Context context) {
+        return "{\"statusCode\": 200}";
+    }
+
+    public class SampleMessageHandler implements SqsMessageHandler<Object> {
+
+        @Override
+        public String process(SQSMessage message) {
+            if("19dd0b57-b21e-4ac1-bd88-01bbb068cb99".equals(message.getMessageId())) {
+                throw new RuntimeException(message.getMessageId());
+            }
+            LOG.info("Processing message with details {}", message);
+            return message.getMessageId();
+        }
+    }
+}

example/HelloWorldFunction/src/main/java/helloworld/AppSqsEventUtil.java

@@ -0,0 +1,39 @@
+package helloworld;
+
+import java.util.List;
+
+import com.amazonaws.services.lambda.runtime.Context;
+import com.amazonaws.services.lambda.runtime.RequestHandler;
+import com.amazonaws.services.lambda.runtime.events.SQSEvent;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import software.amazon.lambda.powertools.sqs.PowertoolsSqs;
+import software.amazon.lambda.powertools.sqs.SQSBatchProcessingException;
+
+import static java.util.Collections.emptyList;
+
+public class AppSqsEventUtil implements RequestHandler<SQSEvent, List<String>> {
+    private static final Logger LOG = LogManager.getLogger(AppSqsEventUtil.class);
+
+    @Override
+    public List<String> handleRequest(SQSEvent input, Context context) {
+        try {
+
+            return PowertoolsSqs.batchProcessor(input, (message) -> {
+                if ("19dd0b57-b21e-4ac1-bd88-01bbb068cb99".equals(message.getMessageId())) {
+                    throw new RuntimeException(message.getMessageId());
+                }
+
+                LOG.info("Processing message with details {}", message);
+                return message.getMessageId();
+            });
+
+        } catch (SQSBatchProcessingException e) {
+            LOG.info("Exception details {}", e.getMessage(), e);
+            LOG.info("Success message Returns{}", e.successMessageReturnValues());
+            LOG.info("Failed messages {}", e.getFailures());
+            LOG.info("Failed messages Reasons {}", e.getExceptions());
+            return emptyList();
+        }
+    }
+}