feat: Add Batch Processor module

.sonarcloud.properties

@@ -0,0 +1,2 @@
+# Ignore code duplicates in the examples
+sonar.cpd.exclusions=examples/**/*,powertools-e2e-tests/**/*

examples/README.md

@@ -9,9 +9,10 @@ Each example can be copied from its subdirectory and used independently of the r
 * [powertools-examples-idempotency](powertools-examples-idempotency) - An idempotent HTTP API
 * [powertools-examples-parameters](powertools-examples-parameters) - Uses the parameters module to provide runtime parameters to a function
 * [powertools-examples-serialization](powertools-examples-serialization) - Uses the serialization module to serialize and deserialize API Gateway & SQS payloads
-* [powertools-examples-sqs](powertools-examples-sqs) - Processes SQS batch requests
+* [powertools-examples-sqs](powertools-examples-sqs) - Processes SQS batch requests (**Deprecated** - will be replaced by `powertools-examples-batch` in version 2 of this library)
 * [powertools-examples-validation](powertools-examples-validation) - Uses the validation module to validate user requests received via API Gateway
 * [powertools-examples-cloudformation](powertools-examples-cloudformation) - Deploys a Cloudformation custom resource
+* [powertools-examples-batch](powertools-examples-batch) - Examples for each of the different batch processing deployments
 
 ## Working with AWS Serverless Application Model (SAM) Examples
 Many of the examples use [AWS Serverless Application Model](https://aws.amazon.com/serverless/sam/) (SAM). To get started

examples/pom.xml

@@ -34,6 +34,7 @@
         <module>powertools-examples-parameters</module>
         <module>powertools-examples-serialization</module>
         <module>powertools-examples-sqs</module>
+        <module>powertools-examples-batch</module>
         <module>powertools-examples-validation</module>
         <module>powertools-examples-cloudformation</module>
     </modules>

examples/powertools-examples-batch/README.md

@@ -0,0 +1,35 @@
+                                         #  Powertools for AWS Lambda (Java) - Batch Example
+
+This project contains examples of Lambda function using the batch processing module of Powertools for AWS Lambda (Java).
+For more information on this module, please refer to the
+[documentation](https://docs.powertools.aws.dev/lambda-java/utilities/batch/).
+
+Three different examples and SAM deployments are included, covering each of the batch sources:
+
+* [SQS](src/main/java/org/demo/batch/sqs) - SQS batch processing
+* [Kinesis Streams](src/main/java/org/demo/batch/kinesis) - Kinesis Streams batch processing
+* [DynamoDB Streams](src/main/java/org/demo/batch/dynamo) - DynamoDB Streams batch processing
+
+## Deploy the sample application
+
+This sample is based on Serverless Application Model (SAM). To deploy it, check out the instructions for getting
+started with SAM in [the examples directory](../README.md)
+
+This sample contains three different deployments, depending on which batch processor you'd like to use, you can
+change to the subdirectory containing the example SAM template, and deploy. For instance, for the SQS batch
+deployment:
+
+```bash
+cd deploy/sqs
+sam build
+sam deploy --guided
+```
+
+## Test the application
+
+Each of the examples uses a Lambda scheduled every 5 minutes to push a batch, and a separate lambda to read it. To
+see this in action, we can simply tail the logs of our stack:
+
+```bash
+sam logs --tail $STACK_NAME
+```

examples/powertools-examples-batch/deploy/ddb-streams/template.yaml

@@ -0,0 +1,74 @@
+AWSTemplateFormatVersion: '2010-09-09'
+Transform: AWS::Serverless-2016-10-31
+Description: >
+  DynamoDB Streams batch processing demo
+
+Globals:
+  Function:
+    Timeout: 20
+    Runtime: java11
+    MemorySize: 512
+    Tracing: Active
+    Architectures:
+      - x86_64
+    Environment:
+      Variables:
+        POWERTOOLS_LOG_LEVEL: INFO
+        POWERTOOLS_LOGGER_SAMPLE_RATE: 1.0
+        POWERTOOLS_LOGGER_LOG_EVENT: true
+
+Resources:
+  DynamoDBTable:
+    Type: AWS::DynamoDB::Table
+    Properties:
+      AttributeDefinitions:
+        - AttributeName: id
+          AttributeType: S
+      KeySchema:
+        - AttributeName: id
+          KeyType: HASH
+      ProvisionedThroughput:
+        ReadCapacityUnits: 5
+        WriteCapacityUnits: 5
+      StreamSpecification:
+        StreamViewType: NEW_IMAGE
+
+
+  DemoDynamoDBWriter:
+    Type: AWS::Serverless::Function
+    Properties:
+      CodeUri: ../..
+      Handler: org.demo.batch.dynamo.DynamoDBWriter::handleRequest
+      Environment:
+        Variables:
+          POWERTOOLS_SERVICE_NAME: ddbstreams-demo
+          TABLE_NAME: !Ref DynamoDBTable
+      Policies:
+        - DynamoDBCrudPolicy:
+            TableName: !Ref DynamoDBTable
+      Events:
+        CWSchedule:
+          Type: Schedule
+          Properties:
+            Schedule: 'rate(1 minute)'
+            Name: !Join [ "-", [ "ddb-writer-schedule", !Select [ 0, !Split [ -, !Select [ 2, !Split [ /, !Ref AWS::StackId ] ] ] ] ] ]
+            Description: Write records to DynamoDB via a Lambda function
+            Enabled: true
+
+  DemoDynamoDBStreamsConsumerFunction:
+    Type: AWS::Serverless::Function
+    Properties:
+      CodeUri: ../..
+      Handler: org.demo.batch.dynamo.DynamoDBStreamBatchHandler::handleRequest
+      Environment:
+        Variables:
+          POWERTOOLS_SERVICE_NAME: ddbstreams-batch-demo
+      Policies: AWSLambdaDynamoDBExecutionRole
+      Events:
+        Stream:
+          Type: DynamoDB
+          Properties:
+            Stream: !GetAtt DynamoDBTable.StreamArn
+            BatchSize: 100
+            StartingPosition: TRIM_HORIZON
+

examples/powertools-examples-batch/deploy/kinesis/template.yml

@@ -0,0 +1,83 @@
+AWSTemplateFormatVersion: '2010-09-09'
+Transform: AWS::Serverless-2016-10-31
+Description: >
+  Kinesis batch processing demo
+
+Globals:
+  Function:
+    Timeout: 20
+    Runtime: java11
+    MemorySize: 512
+    Tracing: Active
+    Environment:
+      Variables:
+        POWERTOOLS_LOG_LEVEL: INFO
+        POWERTOOLS_LOGGER_SAMPLE_RATE: 1.0
+        POWERTOOLS_LOGGER_LOG_EVENT: true
+
+Resources:
+
+  DemoKinesisStream:
+    Type: AWS::Kinesis::Stream
+    Properties:
+      ShardCount: 1
+
+  StreamConsumer:
+    Type: "AWS::Kinesis::StreamConsumer"
+    Properties:
+      StreamARN: !GetAtt DemoKinesisStream.Arn
+      ConsumerName: KinesisBatchHandlerConsumer
+
+  DemoKinesisSenderFunction:
+    Type: AWS::Serverless::Function
+    Properties:
+      CodeUri: ../..
+      Handler: org.demo.batch.kinesis.KinesisBatchSender::handleRequest
+      Environment:
+        Variables:
+          POWERTOOLS_SERVICE_NAME: kinesis-batch-demo
+          STREAM_NAME: !Ref DemoKinesisStream
+      Policies:
+        - Statement:
+            - Sid: WriteToKinesis
+              Effect: Allow
+              Action:
+                - kinesis:PutRecords
+                - kinesis:DescribeStream
+              Resource: !GetAtt DemoKinesisStream.Arn
+      Events:
+        CWSchedule:
+          Type: Schedule
+          Properties:
+            Schedule: 'rate(5 minutes)'
+            Name: !Join [ "-", [ "message-producer-schedule", !Select [ 0, !Split [ -, !Select [ 2, !Split [ /, !Ref AWS::StackId ] ] ] ] ] ]
+            Description: Produce message to Kinesis via a Lambda function
+            Enabled: true
+
+  DemoKinesisConsumerFunction:
+    Type: AWS::Serverless::Function
+    Properties:
+      CodeUri: ../..
+      Handler: org.demo.batch.kinesis.KinesisBatchHandler::handleRequest
+      Environment:
+        Variables:
+          POWERTOOLS_SERVICE_NAME: kinesis-demo
+      Events:
+        Kinesis:
+          Type: Kinesis
+          Properties:
+            Stream: !GetAtt StreamConsumer.ConsumerARN
+            StartingPosition: LATEST
+            BatchSize: 2
+
+Outputs:
+  DemoKinesisQueue:
+    Description: "ARN for Kinesis Stream"
+    Value: !GetAtt DemoKinesisStream.Arn
+  DemoKinesisSenderFunction:
+    Description: "Kinesis Batch Sender - Lambda Function ARN"
+    Value: !GetAtt DemoKinesisSenderFunction.Arn
+  DemoSQSConsumerFunction:
+    Description: "SQS Batch Handler - Lambda Function ARN"
+    Value: !GetAtt DemoKinesisConsumerFunction.Arn
+

examples/powertools-examples-batch/deploy/sqs/template.yml

@@ -0,0 +1,147 @@
+AWSTemplateFormatVersion: '2010-09-09'
+Transform: AWS::Serverless-2016-10-31
+Description: >
+  sqs batch processing demo
+
+Globals:
+  Function:
+    Timeout: 20
+    Runtime: java11
+    MemorySize: 512
+    Tracing: Active
+    Environment:
+      Variables:
+        POWERTOOLS_LOG_LEVEL: INFO
+        POWERTOOLS_LOGGER_SAMPLE_RATE: 1.0
+        POWERTOOLS_LOGGER_LOG_EVENT: true
+
+Resources:
+  CustomerKey:
+    Type: AWS::KMS::Key
+    Properties:
+      Description: KMS key for encrypted queues
+      Enabled: true
+      KeyPolicy:
+        Version: '2012-10-17'
+        Statement:
+          - Sid: Enable IAM User Permissions
+            Effect: Allow
+            Principal:
+              AWS: !Sub 'arn:aws:iam::${AWS::AccountId}:root'
+            Action: 'kms:*'
+            Resource: '*'
+          - Sid: Allow use of the key
+            Effect: Allow
+            Principal:
+              Service: lambda.amazonaws.com
+            Action:
+              - kms:Decrypt
+              - kms:GenerateDataKey
+            Resource: '*'
+
+  CustomerKeyAlias:
+    Type: AWS::KMS::Alias
+    Properties:
+      AliasName: alias/powertools-batch-sqs-demo
+      TargetKeyId: !Ref CustomerKey
+
+  DemoDlqSqsQueue:
+    Type: AWS::SQS::Queue
+    Properties:
+      KmsMasterKeyId: !Ref CustomerKey
+
+  DemoSqsQueue:
+    Type: AWS::SQS::Queue
+    Properties:
+      RedrivePolicy:
+        deadLetterTargetArn:
+          Fn::GetAtt:
+            - "DemoDlqSqsQueue"
+            - "Arn"
+        maxReceiveCount: 2
+      KmsMasterKeyId: !Ref CustomerKey
+
+  DemoSQSSenderFunction:
+    Type: AWS::Serverless::Function
+    Properties:
+      CodeUri: ../..
+      Handler: org.demo.batch.sqs.SqsBatchSender::handleRequest
+      Environment:
+        Variables:
+          POWERTOOLS_SERVICE_NAME: sqs-batch-demo
+          QUEUE_URL: !Ref DemoSqsQueue
+      Policies:
+        - Statement:
+            - Sid: SQSSendMessageBatch
+              Effect: Allow
+              Action:
+                - sqs:SendMessageBatch
+                - sqs:SendMessage
+              Resource: !GetAtt DemoSqsQueue.Arn
+            - Sid: SQSKMSKey
+              Effect: Allow
+              Action:
+                - kms:GenerateDataKey
+                - kms:Decrypt
+              Resource: !GetAtt CustomerKey.Arn
+      Events:
+        CWSchedule:
+          Type: Schedule
+          Properties:
+            Schedule: 'rate(5 minutes)'
+            Name: !Join [ "-", [ "message-producer-schedule", !Select [ 0, !Split [ -, !Select [ 2, !Split [ /, !Ref AWS::StackId ] ] ] ] ] ]
+            Description: Produce message to SQS via a Lambda function
+            Enabled: true
+
+  DemoSQSConsumerFunction:
+    Type: AWS::Serverless::Function
+    Properties:
+      CodeUri: ../..
+      Handler: org.demo.batch.sqs.SqsBatchHandler::handleRequest
+      Environment:
+        Variables:
+          POWERTOOLS_SERVICE_NAME: sqs-demo
+      Policies:
+        - Statement:
+            - Sid: SQSDeleteGetAttribute
+              Effect: Allow
+              Action:
+                - sqs:DeleteMessageBatch
+                - sqs:GetQueueAttributes
+              Resource: !GetAtt DemoSqsQueue.Arn
+            - Sid: SQSSendMessageBatch
+              Effect: Allow
+              Action:
+                - sqs:SendMessageBatch
+                - sqs:SendMessage
+              Resource: !GetAtt DemoDlqSqsQueue.Arn
+            - Sid: SQSKMSKey
+              Effect: Allow
+              Action:
+                - kms:GenerateDataKey
+                - kms:Decrypt
+              Resource: !GetAtt CustomerKey.Arn
+      Events:
+        MySQSEvent:
+          Type: SQS
+          Properties:
+            Queue: !GetAtt DemoSqsQueue.Arn
+            BatchSize: 2
+            MaximumBatchingWindowInSeconds: 300
+
+Outputs:
+  DemoSqsQueue:
+    Description: "ARN for main SQS queue"
+    Value: !GetAtt DemoSqsQueue.Arn
+  DemoDlqSqsQueue:
+    Description: "ARN for DLQ"
+    Value: !GetAtt DemoDlqSqsQueue.Arn
+  DemoSQSSenderFunction:
+    Description: "SQS Batch Sender - Lambda Function ARN"
+    Value: !GetAtt DemoSQSSenderFunction.Arn
+  DemoSQSConsumerFunction:
+    Description: "SQS Batch Handler - Lambda Function ARN"
+    Value: !GetAtt DemoSQSConsumerFunction.Arn
+  DemoSQSConsumerFunctionRole:
+    Description: "Implicit IAM Role created for SQS Lambda Function ARN"
+    Value: !GetAtt DemoSQSConsumerFunctionRole.Arn