DynamoDB E2E works

scottgerring · scottgerring · commit 006b9e0a7cc5 · 2023-08-03T19:41:46.000+02:00
diff --git a/powertools-batch/src/main/java/software/amazon/lambda/powertools/batch/handler/DynamoDbBatchMessageHandler.java b/powertools-batch/src/main/java/software/amazon/lambda/powertools/batch/handler/DynamoDbBatchMessageHandler.java
@@ -60,6 +60,7 @@ public StreamsEventResponse processBatch(DynamodbEvent event, Context context) {
                 String sequenceNumber = record.getDynamodb().getSequenceNumber();
                 LOGGER.error("Error while processing record with id {}: {}, adding it to batch item failures",
                         sequenceNumber, t.getMessage());
+                LOGGER.error("Error was", t);
                 batchFailures.add(new StreamsEventResponse.BatchItemFailure(sequenceNumber));
 
                 // Report failure if we have a handler
diff --git a/powertools-batch/src/main/java/software/amazon/lambda/powertools/batch/handler/KinesisStreamsBatchMessageHandler.java b/powertools-batch/src/main/java/software/amazon/lambda/powertools/batch/handler/KinesisStreamsBatchMessageHandler.java
@@ -73,6 +73,11 @@ public StreamsEventResponse processBatch(KinesisEvent event, Context context) {
                     this.successHandler.accept(record);
                 }
             } catch (Throwable t) {
+                String sequenceNumber = record.getEventID();
+                LOGGER.error("Error while processing record with eventID {}: {}, adding it to batch item failures",
+                        sequenceNumber, t.getMessage());
+                LOGGER.error("Error was", t);
+
                 batchFailures.add(new StreamsEventResponse.BatchItemFailure(record.getKinesis().getSequenceNumber()));
 
                 // Report failure if we have a handler
diff --git a/powertools-batch/src/main/java/software/amazon/lambda/powertools/batch/handler/SqsBatchMessageHandler.java b/powertools-batch/src/main/java/software/amazon/lambda/powertools/batch/handler/SqsBatchMessageHandler.java
@@ -86,6 +86,8 @@ public SQSBatchResponse processBatch(SQSEvent event, Context context) {
             } catch (Throwable t) {
                 LOGGER.error("Error while processing message with messageId {}: {}, adding it to batch item failures",
                         message.getMessageId(), t.getMessage());
+                LOGGER.error("Error was", t);
+
                 response.getBatchItemFailures()
                         .add(SQSBatchResponse.BatchItemFailure.builder().withItemIdentifier(message.getMessageId())
                                 .build());
diff --git a/powertools-e2e-tests/handlers/batch/src/main/java/software/amazon/lambda/powertools/e2e/Function.java b/powertools-e2e-tests/handlers/batch/src/main/java/software/amazon/lambda/powertools/e2e/Function.java
@@ -37,6 +37,7 @@
 import java.io.OutputStreamWriter;
 import java.nio.charset.StandardCharsets;
 import java.util.HashMap;
+import java.util.Map;
 import java.util.stream.Collectors;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
@@ -50,6 +51,8 @@
 import software.amazon.lambda.powertools.logging.Logging;
 import software.amazon.lambda.powertools.utilities.JsonConfig;
 
+import javax.management.Attribute;
+
 
 public class Function implements RequestHandler<InputStream, Object> {
 
@@ -83,30 +86,46 @@ private void processProductMessage(Product p, Context c) {
         // TODO - write product details to output table
         ddbClient = DynamoDbClient.builder()
                 .build();
+        Map<String, AttributeValue> results = new HashMap<>();
+        results.put("functionName", AttributeValue.builder()
+                .s(c.getFunctionName())
+                .build());
+        results.put("id", AttributeValue.builder()
+            .s(Long.toString(p.getId()))
+            .build());
+        results.put("name", AttributeValue.builder()
+                .s(p.getName())
+                .build());
+        results.put("price", AttributeValue.builder()
+                .n(Double.toString(p.getPrice()))
+                .build());
         ddbClient.putItem(PutItemRequest.builder()
                         .tableName(ddbOutputTable)
-                        .item(new HashMap<String, AttributeValue>() {
-                            {
-                                put("functionName", AttributeValue.builder()
-                                        .s(c.getFunctionName())
-                                        .build());
-                                put("id", AttributeValue.builder()
-                                        .s(Long.toString(p.getId()))
-                                        .build());
-                                put("name", AttributeValue.builder()
-                                        .s(p.getName())
-                                        .build());
-                                put("price", AttributeValue.builder()
-                                        .n(Double.toString(p.getPrice()))
-                                        .build());
-                            }})
+                        .item(results)
                 .build());
     }
 
     private void processDdbMessage(DynamodbEvent.DynamodbStreamRecord dynamodbStreamRecord, Context context) {
         LOGGER.info("Processing DynamoDB Stream Record" + dynamodbStreamRecord);
 
-        // TODO write DDB change details to batch output table
+        ddbClient = DynamoDbClient.builder()
+                .build();
+
+        String id = dynamodbStreamRecord.getDynamodb().getKeys().get("id").getS();
+        LOGGER.info("Incoming ID is " + id);
+
+        Map<String, AttributeValue> results = new HashMap<>();
+        results.put("functionName", AttributeValue.builder()
+                .s(context.getFunctionName())
+                .build());
+        results.put("id", AttributeValue.builder()
+                .s(id)
+                .build());
+
+        ddbClient.putItem(PutItemRequest.builder()
+                .tableName(ddbOutputTable)
+                .item(results)
+                .build());
     }
 
     public Object createResult(String input, Context context) {
diff --git a/powertools-e2e-tests/handlers/batch/src/test/java/TestSerialization.java b/powertools-e2e-tests/handlers/batch/src/test/java/TestSerialization.java
diff --git a/powertools-e2e-tests/src/test/java/software/amazon/lambda/powertools/BatchE2ET.java b/powertools-e2e-tests/src/test/java/software/amazon/lambda/powertools/BatchE2ET.java
@@ -42,10 +42,7 @@
 import software.amazon.awssdk.regions.Region;
 import software.amazon.awssdk.services.dynamodb.DynamoDbClient;
 import software.amazon.awssdk.services.dynamodb.DynamoDbClientBuilder;
-import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
-import software.amazon.awssdk.services.dynamodb.model.DeleteItemRequest;
-import software.amazon.awssdk.services.dynamodb.model.ScanRequest;
-import software.amazon.awssdk.services.dynamodb.model.ScanResponse;
+import software.amazon.awssdk.services.dynamodb.model.*;
 import software.amazon.awssdk.services.kinesis.KinesisClient;
 import software.amazon.awssdk.services.kinesis.model.PutRecordsRequest;
 import software.amazon.awssdk.services.kinesis.model.PutRecordsRequestEntry;
@@ -72,6 +69,7 @@ public class BatchE2ET {
     private static DynamoDbClient ddbClient;
     private static SqsClient sqsClient;
     private static KinesisClient kinesisClient;
+    private static String ddbStreamsTestTable;
     private final List<Product> testProducts;
 
     @BeforeAll
@@ -80,13 +78,16 @@ public static void setup() {
         String random = UUID.randomUUID().toString().substring(0, 6);
         String queueName = "batchqueue" + random;
         kinesisStreamName = "batchstream" + random;
+        ddbStreamsTestTable = "ddbstreams" + random;
+
         objectMapper = JsonConfig.get().getObjectMapper();
 
         infrastructure = Infrastructure.builder()
                 .testName(BatchE2ET.class.getSimpleName())
                 .pathToFunction("batch")
                 .idempotencyTable("idempo" + random)
                 .queue(queueName)
+                .ddbStreamsTableName(ddbStreamsTestTable)
                 .kinesisStream(kinesisStreamName)
                 .build();
 
@@ -95,6 +96,7 @@ public static void setup() {
         queueUrl = outputs.get("QueueURL");
         kinesisStreamName = outputs.get("KinesisStreamName");
         outputTable = outputs.get("TableNameForAsyncTests");
+        ddbStreamsTestTable = outputs.get("DdbStreamsTestTable");
         
         ddbClient = DynamoDbClient.builder()
                 .region(region)
@@ -211,6 +213,32 @@ public void kinesisBatchProcessingSucceeds() throws InterruptedException {
         validateAllItemsHandled(items);
     }
 
+    @Test
+    public void ddbStreamsBatchProcessingSucceeds() throws InterruptedException {
+        // GIVEN
+        String theId = "my-test-id";
+
+        // WHEN
+       ddbClient.putItem(PutItemRequest.builder()
+               .tableName(ddbStreamsTestTable)
+               .item(new HashMap<String, AttributeValue>() {
+                   {
+                       put("id", AttributeValue.builder()
+                               .s(theId)
+                               .build());
+                   }})
+               .build());
+        Thread.sleep(90000); // wait for function to be executed
+
+        // THEN
+        ScanResponse items = ddbClient.scan(ScanRequest.builder()
+                .tableName(outputTable)
+                .build());
+
+        assertThat(items.count()).isEqualTo(1);
+        assertThat(items.items().get(0).get("id").s()).isEqualTo(theId);
+    }
+
     private void validateAllItemsHandled(ScanResponse items) {
         for (Product p : testProducts) {
             boolean foundIt = false;
diff --git a/powertools-e2e-tests/src/test/java/software/amazon/lambda/powertools/testutils/Infrastructure.java b/powertools-e2e-tests/src/test/java/software/amazon/lambda/powertools/testutils/Infrastructure.java
@@ -16,7 +16,6 @@
 
 import static java.util.Collections.singletonList;
 
-import com.amazonaws.services.lambda.runtime.events.KinesisEvent;
 import com.fasterxml.jackson.databind.JsonNode;
 import java.io.File;
 import java.io.IOException;
@@ -46,18 +45,15 @@
 import software.amazon.awscdk.services.appconfig.CfnDeploymentStrategy;
 import software.amazon.awscdk.services.appconfig.CfnEnvironment;
 import software.amazon.awscdk.services.appconfig.CfnHostedConfigurationVersion;
-import software.amazon.awscdk.services.dynamodb.Attribute;
-import software.amazon.awscdk.services.dynamodb.AttributeType;
-import software.amazon.awscdk.services.dynamodb.BillingMode;
-import software.amazon.awscdk.services.dynamodb.Table;
-import software.amazon.awscdk.services.events.targets.KinesisStream;
+import software.amazon.awscdk.services.dynamodb.*;
 import software.amazon.awscdk.services.iam.PolicyStatement;
 import software.amazon.awscdk.services.kinesis.Stream;
 import software.amazon.awscdk.services.kinesis.StreamMode;
 import software.amazon.awscdk.services.lambda.Code;
 import software.amazon.awscdk.services.lambda.Function;
 import software.amazon.awscdk.services.lambda.StartingPosition;
 import software.amazon.awscdk.services.lambda.Tracing;
+import software.amazon.awscdk.services.lambda.eventsources.DynamoEventSource;
 import software.amazon.awscdk.services.lambda.eventsources.KinesisEventSource;
 import software.amazon.awscdk.services.lambda.eventsources.SqsEventSource;
 import software.amazon.awscdk.services.logs.LogGroup;
@@ -118,6 +114,7 @@ public class Infrastructure {
     private final String queue;
     private final String kinesisStream;
     private final String largeMessagesBucket;
+    private String ddbStreamsTableName;
 
     private String functionName;
     private Object cfnTemplate;
@@ -135,6 +132,7 @@ private Infrastructure(Builder builder) {
         this.queue = builder.queue;
         this.kinesisStream = builder.kinesisStream;
         this.largeMessagesBucket = builder.largeMessagesBucket;
+        this.ddbStreamsTableName = builder.ddbStreamsTableName;
 
         this.app = new App();
         this.stack = createStackWithLambda();
@@ -317,6 +315,23 @@ private Stack createStackWithLambda() {
                     .build();
         }
 
+        if (!StringUtils.isEmpty(ddbStreamsTableName)) {
+            Table ddbStreamsTable = Table.Builder.create(stack, "DDBStreamsTable")
+                    .tableName(ddbStreamsTableName)
+                    .stream(StreamViewType.KEYS_ONLY)
+                    .removalPolicy(RemovalPolicy.DESTROY)
+                    .partitionKey(Attribute.builder().name("id").type(AttributeType.STRING).build())
+                    .build();
+
+            DynamoEventSource ddbEventSource = DynamoEventSource.Builder.create(ddbStreamsTable)
+                    .batchSize(1)
+                    .startingPosition(StartingPosition.TRIM_HORIZON)
+                    .maxBatchingWindow(Duration.seconds(1))
+                    .build();
+            function.addEventSource(ddbEventSource);
+            CfnOutput.Builder.create(stack, "DdbStreamsTestTable").value(ddbStreamsTable.getTableName()).build();
+        }
+
         if (!StringUtils.isEmpty(largeMessagesBucket)) {
             Bucket offloadBucket = Bucket.Builder
                     .create(stack, "LargeMessagesOffloadBucket")
@@ -481,6 +496,7 @@ public static class Builder {
         private String idemPotencyTable;
         private String queue;
         private String kinesisStream;
+        private String ddbStreamsTableName;
 
         private Builder() {
             getJavaRuntime();
@@ -561,6 +577,11 @@ public Builder kinesisStream(String stream) {
             return this;
         }
 
+        public Builder ddbStreamsTableName(String tableName) {
+            this.ddbStreamsTableName = tableName;
+            return this;
+        }
+
         public Builder largeMessagesBucket(String largeMessagesBucket) {
             this.largeMessagesBucket = largeMessagesBucket;
             return this;
