Better error handling

Author: Pankaj Agrawal

powertools-sqs/src/main/java/software/amazon/lambda/powertools/sqs/PowertoolsSqs.java

@@ -14,28 +14,30 @@
 package software.amazon.lambda.powertools.sqs;
 
 import java.util.ArrayList;
-import java.util.Collections;
 import java.util.List;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
 import com.amazonaws.services.lambda.runtime.events.SQSEvent;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import software.amazon.awssdk.services.sqs.SqsClient;
 import software.amazon.lambda.powertools.sqs.internal.BatchContext;
-import software.amazon.lambda.powertools.sqs.internal.SqsMessageAspect;
+import software.amazon.lambda.powertools.sqs.internal.SqsLargeMessageAspect;
 import software.amazon.payloadoffloading.PayloadS3Pointer;
 
 import static com.amazonaws.services.lambda.runtime.events.SQSEvent.SQSMessage;
-import static software.amazon.lambda.powertools.sqs.internal.SqsMessageAspect.processMessages;
+import static software.amazon.lambda.powertools.sqs.internal.SqsLargeMessageAspect.processMessages;
 
 /**
  * A class of helper functions to add additional functionality to LargeMessageHandler.
  * <p>
  * {@see PowertoolsLogging}
  */
 public final class PowertoolsSqs {
+    private static final Log LOG = LogFactory.getLog(PowertoolsSqs.class);
 
     private static final ObjectMapper objectMapper = new ObjectMapper();
     private static SqsClient client = SqsClient.create();
@@ -79,7 +81,7 @@ public static <R> R enrichedMessageFromS3(final SQSEvent sqsEvent,
         R returnValue = messageFunction.apply(sqsMessages);
 
         if (deleteS3Payload) {
-            s3Pointers.forEach(SqsMessageAspect::deleteMessage);
+            s3Pointers.forEach(SqsLargeMessageAspect::deleteMessage);
         }
 
         return returnValue;
@@ -94,14 +96,15 @@ public static SqsClient defaultSqsClient() {
     }
 
     public static <R> List<R> partialBatchProcessor(final SQSEvent event,
-                                                    boolean suppressException,
+                                                    final boolean suppressException,
                                                     final Class<? extends SqsMessageHandler<R>> handler) {
 
         try {
             return partialBatchProcessor(event, suppressException, handler.newInstance());
         } catch (IllegalAccessException | InstantiationException e) {
-            // LOG something
-            return Collections.emptyList();
+            LOG.error("Failed invoking process method on handler", e);
+            throw new RuntimeException("Unexpected error occurred. Please raise issue at " +
+                    "https://github.com/awslabs/aws-lambda-powertools-java/issues", e);
         }
     }
 
@@ -134,5 +137,4 @@ private static SQSMessage clonedMessage(final SQSMessage sqsMessage) {
             throw new RuntimeException(e);
         }
     }
-
 }

powertools-sqs/src/main/java/software/amazon/lambda/powertools/sqs/internal/BatchContext.java

@@ -3,48 +3,73 @@
 import java.util.ArrayList;
 import java.util.List;
 
-import com.amazonaws.services.lambda.runtime.events.SQSEvent;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import software.amazon.awssdk.services.sqs.SqsClient;
 import software.amazon.awssdk.services.sqs.model.DeleteMessageBatchRequest;
 import software.amazon.awssdk.services.sqs.model.DeleteMessageBatchRequestEntry;
 import software.amazon.awssdk.services.sqs.model.GetQueueUrlRequest;
 import software.amazon.lambda.powertools.sqs.SQSBatchProcessingException;
 
+import static com.amazonaws.services.lambda.runtime.events.SQSEvent.SQSMessage;
+import static java.lang.String.format;
 import static java.util.stream.Collectors.toList;
 
 public final class BatchContext {
-    private List<SQSEvent.SQSMessage> success = new ArrayList<>();
-    private List<SQSEvent.SQSMessage> failures = new ArrayList<>();
-    private List<Exception> exceptions = new ArrayList<>();
-    private SqsClient client;
+    private static final Log LOG = LogFactory.getLog(BatchContext.class);
+
+    private final List<SQSMessage> success = new ArrayList<>();
+    private final List<SQSMessage> failures = new ArrayList<>();
+    private final List<Exception> exceptions = new ArrayList<>();
+    private final SqsClient client;
 
     public BatchContext(SqsClient client) {
         this.client = client;
     }
 
-    public void addSuccess(SQSEvent.SQSMessage event) {
+    public void addSuccess(SQSMessage event) {
         success.add(event);
     }
 
-    public void addFailure(SQSEvent.SQSMessage event, Exception e) {
+    public void addFailure(SQSMessage event, Exception e) {
         failures.add(event);
         exceptions.add(e);
     }
 
+    public void processSuccessAndReset(final boolean suppressException) {
+        try {
+            if (hasFailures()) {
+
+                deleteSuccessMessage();
+
+                if (suppressException) {
+                    List<String> messageIds = failures.stream().map(SQSMessage::getMessageId).collect(toList());
+                    LOG.debug(format("[%s] records failed processing, but exceptions are suppressed. Failed messages %s", failures.size(), messageIds));
+                } else {
+                    throw new SQSBatchProcessingException(exceptions);
+                }
+            }
+        } finally {
+            reset();
+        }
+    }
+
     private boolean hasFailures() {
         return !failures.isEmpty();
     }
 
-    private void cleanUpAndReset() {
-        DeleteMessageBatchRequest request = DeleteMessageBatchRequest.builder()
-                .queueUrl(url())
-                .entries(success.stream().map(m -> DeleteMessageBatchRequestEntry.builder()
-                        .id(m.getMessageId())
-                        .receiptHandle(m.getReceiptHandle())
-                        .build()).collect(toList()))
-                .build();
+    private void deleteSuccessMessage() {
+        if (!success.isEmpty()) {
+            DeleteMessageBatchRequest request = DeleteMessageBatchRequest.builder()
+                    .queueUrl(url())
+                    .entries(success.stream().map(m -> DeleteMessageBatchRequestEntry.builder()
+                            .id(m.getMessageId())
+                            .receiptHandle(m.getReceiptHandle())
+                            .build()).collect(toList()))
+                    .build();
 
-        client.deleteMessageBatch(request);
+            client.deleteMessageBatch(request);
+        }
     }
 
     private String url() {
@@ -57,21 +82,8 @@ private String url() {
     }
 
     private void reset() {
-        success = new ArrayList<>();
-        failures = new ArrayList<>();
-        exceptions = new ArrayList<>();
-    }
-
-    public void processSuccessAndReset(final boolean suppressException) {
-        if (hasFailures() && !suppressException) {
-            SQSBatchProcessingException exception = new SQSBatchProcessingException(exceptions);
-            cleanUpAndReset();
-            throw exception;
-        } else if (hasFailures()) {
-            // LOG
-            cleanUpAndReset();
-        } else {
-            reset();
-        }
+        success.clear();
+        failures.clear();
+        exceptions.clear();
     }
 }

powertools-sqs/src/main/java/software/amazon/lambda/powertools/sqs/internal/SqsMessageAspect.java renamed to powertools-sqs/src/main/java/software/amazon/lambda/powertools/sqs/internal/SqsLargeMessageAspect.java

@@ -28,9 +28,9 @@
 import static software.amazon.lambda.powertools.core.internal.LambdaHandlerProcessor.isHandlerMethod;
 
 @Aspect
-public class SqsMessageAspect {
+public class SqsLargeMessageAspect {
 
-    private static final Log LOG = LogFactory.getLog(SqsMessageAspect.class);
+    private static final Log LOG = LogFactory.getLog(SqsLargeMessageAspect.class);
     private static AmazonS3 amazonS3 = AmazonS3ClientBuilder.defaultClient();
 
     @SuppressWarnings({"EmptyMethod"})

powertools-sqs/src/main/java/software/amazon/lambda/powertools/sqs/internal/SqsMessageBatchProcessorAspect.java

@@ -5,17 +5,14 @@
 import org.aspectj.lang.annotation.Around;
 import org.aspectj.lang.annotation.Aspect;
 import org.aspectj.lang.annotation.Pointcut;
-import software.amazon.awssdk.services.sqs.SqsClient;
-import software.amazon.lambda.powertools.sqs.PowertoolsSqs;
 import software.amazon.lambda.powertools.sqs.SqsBatchProcessor;
 
 import static software.amazon.lambda.powertools.core.internal.LambdaHandlerProcessor.isHandlerMethod;
-import static software.amazon.lambda.powertools.sqs.internal.SqsMessageAspect.placedOnSqsEventRequestHandler;
+import static software.amazon.lambda.powertools.sqs.PowertoolsSqs.partialBatchProcessor;
+import static software.amazon.lambda.powertools.sqs.internal.SqsLargeMessageAspect.placedOnSqsEventRequestHandler;
 
 @Aspect
 public class SqsMessageBatchProcessorAspect {
-    private static final SqsClient client = SqsClient.create();
-    private static BatchContext details = new BatchContext(PowertoolsSqs.defaultSqsClient());
 
     @SuppressWarnings({"EmptyMethod"})
     @Pointcut("@annotation(sqsBatchProcessor)")
@@ -32,7 +29,7 @@ && placedOnSqsEventRequestHandler(pjp)) {
 
             SQSEvent sqsEvent = (SQSEvent) proceedArgs[0];
 
-            PowertoolsSqs.partialBatchProcessor(sqsEvent, sqsBatchProcessor.suppressException(), sqsBatchProcessor.value().newInstance());
+            partialBatchProcessor(sqsEvent, sqsBatchProcessor.suppressException(), sqsBatchProcessor.value().newInstance());
         }
 
         return pjp.proceed(proceedArgs);

powertools-sqs/src/test/java/software/amazon/lambda/powertools/sqs/PowertoolsSqsTest.java

@@ -21,7 +21,7 @@
 import org.junit.jupiter.params.provider.MethodSource;
 import org.junit.jupiter.params.provider.ValueSource;
 import org.mockito.Mock;
-import software.amazon.lambda.powertools.sqs.internal.SqsMessageAspect;
+import software.amazon.lambda.powertools.sqs.internal.SqsLargeMessageAspect;
 
 import static com.amazonaws.services.lambda.runtime.events.SQSEvent.SQSMessage;
 import static java.util.Collections.singletonList;
@@ -33,7 +33,7 @@
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoInteractions;
 import static org.mockito.Mockito.when;
-import static org.mockito.MockitoAnnotations.initMocks;
+import static org.mockito.MockitoAnnotations.openMocks;
 
 class PowertoolsSqsTest {
 
@@ -44,8 +44,8 @@ class PowertoolsSqsTest {
 
     @BeforeEach
     void setUp() throws IllegalAccessException {
-        initMocks(this);
-        writeStaticField(SqsMessageAspect.class, "amazonS3", amazonS3, true);
+        openMocks(this);
+        writeStaticField(SqsLargeMessageAspect.class, "amazonS3", amazonS3, true);
     }
 
     @Test
@@ -122,7 +122,7 @@ public void shouldFailEntireBatchIfFailedDownloadingFromS3(RuntimeException exce
         String messageBody = "[\"software.amazon.payloadoffloading.PayloadS3Pointer\",{\"s3BucketName\":\"" + BUCKET_NAME + "\",\"s3Key\":\"" + BUCKET_KEY + "\"}]";
         SQSEvent sqsEvent = messageWithBody(messageBody);
 
-        assertThatExceptionOfType(SqsMessageAspect.FailedProcessingLargePayloadException.class)
+        assertThatExceptionOfType(SqsLargeMessageAspect.FailedProcessingLargePayloadException.class)
                 .isThrownBy(() -> PowertoolsSqs.enrichedMessageFromS3(sqsEvent, sqsMessages -> sqsMessages.get(0).getBody()))
                 .withCause(exception);
 
@@ -145,7 +145,7 @@ public void close() throws IOException {
         String messageBody = "[\"software.amazon.payloadoffloading.PayloadS3Pointer\",{\"s3BucketName\":\"" + BUCKET_NAME + "\",\"s3Key\":\"" + BUCKET_KEY + "\"}]";
         SQSEvent sqsEvent = messageWithBody(messageBody);
 
-        assertThatExceptionOfType(SqsMessageAspect.FailedProcessingLargePayloadException.class)
+        assertThatExceptionOfType(SqsLargeMessageAspect.FailedProcessingLargePayloadException.class)
                 .isThrownBy(() -> PowertoolsSqs.enrichedMessageFromS3(sqsEvent, sqsMessages -> sqsMessages.get(0).getBody()))
                 .withCauseInstanceOf(IOException.class);
 

powertools-sqs/src/test/java/software/amazon/lambda/powertools/sqs/handlers/PartialBatchHandler.java

@@ -12,7 +12,6 @@
 
 public class PartialBatchHandler implements RequestHandler<SQSEvent, List<Object>> {
 
-
     @Override
     @LargeMessageHandler
     @SqsBatchProcessor(HandlerSqs.class)

powertools-sqs/src/test/java/software/amazon/lambda/powertools/sqs/internal/SqsMessageAspectTest.java renamed to powertools-sqs/src/test/java/software/amazon/lambda/powertools/sqs/internal/SqsLargeMessageAspectTest.java

@@ -1,5 +1,9 @@
 package software.amazon.lambda.powertools.sqs.internal;
 
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.util.stream.Stream;
+
 import com.amazonaws.AmazonServiceException;
 import com.amazonaws.SdkClientException;
 import com.amazonaws.services.lambda.runtime.Context;
@@ -21,10 +25,6 @@
 import software.amazon.lambda.powertools.sqs.handlers.SqsMessageHandler;
 import software.amazon.lambda.powertools.sqs.handlers.SqsNoDeleteMessageHandler;
 
-import java.io.ByteArrayInputStream;
-import java.io.IOException;
-import java.util.stream.Stream;
-
 import static com.amazonaws.services.lambda.runtime.events.SQSEvent.SQSMessage;
 import static java.util.Collections.singletonList;
 import static org.apache.commons.lang3.reflect.FieldUtils.writeStaticField;
@@ -35,10 +35,10 @@
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoInteractions;
 import static org.mockito.Mockito.when;
-import static org.mockito.MockitoAnnotations.initMocks;
-import static software.amazon.lambda.powertools.sqs.internal.SqsMessageAspect.FailedProcessingLargePayloadException;
+import static org.mockito.MockitoAnnotations.openMocks;
+import static software.amazon.lambda.powertools.sqs.internal.SqsLargeMessageAspect.FailedProcessingLargePayloadException;
 
-public class SqsMessageAspectTest {
+public class SqsLargeMessageAspectTest {
 
     private RequestHandler<SQSEvent, String> requestHandler;
 
@@ -53,9 +53,9 @@ public class SqsMessageAspectTest {
 
     @BeforeEach
     void setUp() throws IllegalAccessException {
-        initMocks(this);
+        openMocks(this);
         setupContext();
-        writeStaticField(SqsMessageAspect.class, "amazonS3", amazonS3, true);
+        writeStaticField(SqsLargeMessageAspect.class, "amazonS3", amazonS3, true);
         requestHandler = new SqsMessageHandler();
     }