validation of sqs / kinesis batches with partial failures

Author: jeromevdl

powertools-validation/pom.xml

@@ -83,7 +83,21 @@
             <artifactId>junit-jupiter-engine</artifactId>
             <scope>test</scope>
         </dependency>
-
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-simple</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-core</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.amazonaws</groupId>
+            <artifactId>aws-lambda-java-tests</artifactId>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.apache.commons</groupId>
             <artifactId>commons-lang3</artifactId>

powertools-validation/src/main/java/software/amazon/lambda/powertools/validation/internal/ValidationAspect.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2024 Amazon.com, Inc. or its affiliates.
+ * Licensed under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package software.amazon.lambda.powertools.validation.handlers;
+
+import com.amazonaws.services.lambda.runtime.Context;
+import com.amazonaws.services.lambda.runtime.RequestHandler;
+import com.amazonaws.services.lambda.runtime.events.KinesisEvent;
+import com.amazonaws.services.lambda.runtime.events.StreamsEventResponse;
+import java.util.ArrayList;
+import software.amazon.lambda.powertools.validation.Validation;
+
+public class KinesisHandlerWithError implements RequestHandler<KinesisEvent, StreamsEventResponse> {
+
+    @Override
+    @Validation(inboundSchema = "classpath:/schema_v7.json")
+    public StreamsEventResponse handleRequest(KinesisEvent input, Context context) {
+        StreamsEventResponse response = StreamsEventResponse.builder().withBatchItemFailures(new ArrayList<>()).build();
+        assert input.getRecords().size() == 2; // invalid messages have been removed from the input
+        response.getBatchItemFailures().add(StreamsEventResponse.BatchItemFailure.builder().withItemIdentifier("1234").build());
+        return response;
+    }
+}

powertools-validation/src/test/java/software/amazon/lambda/powertools/validation/handlers/KinesisHandlerWithError.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2024 Amazon.com, Inc. or its affiliates.
+ * Licensed under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package software.amazon.lambda.powertools.validation.handlers;
+
+import com.amazonaws.services.lambda.runtime.Context;
+import com.amazonaws.services.lambda.runtime.RequestHandler;
+import com.amazonaws.services.lambda.runtime.events.SQSBatchResponse;
+import com.amazonaws.services.lambda.runtime.events.SQSEvent;
+import java.util.ArrayList;
+import software.amazon.lambda.powertools.validation.Validation;
+
+public class SQSHandlerWithError implements RequestHandler<SQSEvent, SQSBatchResponse> {
+
+    @Override
+    @Validation(inboundSchema = "classpath:/schema_v7.json")
+    public SQSBatchResponse handleRequest(SQSEvent input, Context context) {
+        SQSBatchResponse response = SQSBatchResponse.builder().withBatchItemFailures(new ArrayList<>()).build();
+        assert input.getRecords().size() == 2; // invalid messages have been removed from the input
+        response.getBatchItemFailures().add(SQSBatchResponse.BatchItemFailure.builder().withItemIdentifier("1234").build());
+        return response;
+    }
+}

powertools-validation/src/test/java/software/amazon/lambda/powertools/validation/handlers/SQSHandlerWithError.java

@@ -0,0 +1,32 @@
+/*
+ * Copyright 2024 Amazon.com, Inc. or its affiliates.
+ * Licensed under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package software.amazon.lambda.powertools.validation.handlers;
+
+import com.amazonaws.services.lambda.runtime.Context;
+import com.amazonaws.services.lambda.runtime.RequestHandler;
+import com.amazonaws.services.lambda.runtime.events.KinesisEvent;
+import com.amazonaws.services.lambda.runtime.events.StreamsEventResponse;
+import software.amazon.lambda.powertools.validation.Validation;
+
+public class StandardKinesisHandler implements RequestHandler<KinesisEvent, StreamsEventResponse> {
+
+    @Override
+    @Validation(inboundSchema = "classpath:/schema_v7.json")
+    public StreamsEventResponse handleRequest(KinesisEvent input, Context context) {
+        StreamsEventResponse response = StreamsEventResponse.builder().build();
+        assert input.getRecords().size() == 2; // invalid messages have been removed from the input
+        return response;
+    }
+}

powertools-validation/src/test/java/software/amazon/lambda/powertools/validation/handlers/StandardKinesisHandler.java

@@ -0,0 +1,32 @@
+/*
+ * Copyright 2024 Amazon.com, Inc. or its affiliates.
+ * Licensed under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package software.amazon.lambda.powertools.validation.handlers;
+
+import com.amazonaws.services.lambda.runtime.Context;
+import com.amazonaws.services.lambda.runtime.RequestHandler;
+import com.amazonaws.services.lambda.runtime.events.SQSBatchResponse;
+import com.amazonaws.services.lambda.runtime.events.SQSEvent;
+import software.amazon.lambda.powertools.validation.Validation;
+
+public class StandardSQSHandler implements RequestHandler<SQSEvent, SQSBatchResponse> {
+
+    @Override
+    @Validation(inboundSchema = "classpath:/schema_v7.json")
+    public SQSBatchResponse handleRequest(SQSEvent input, Context context) {
+        SQSBatchResponse response = SQSBatchResponse.builder().build();
+        assert input.getRecords().size() == 2; // invalid messages have been removed from the input
+        return response;
+    }
+}

powertools-validation/src/test/java/software/amazon/lambda/powertools/validation/handlers/StandardSQSHandler.java

@@ -20,24 +20,6 @@
 import static org.junit.jupiter.api.Assertions.fail;
 import static org.mockito.Mockito.when;
 
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.stream.Stream;
-
-import org.aspectj.lang.ProceedingJoinPoint;
-import org.aspectj.lang.Signature;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.Arguments;
-import org.junit.jupiter.params.provider.ArgumentsSource;
-import org.junit.jupiter.params.provider.MethodSource;
-import org.mockito.Mock;
-import org.mockito.MockitoAnnotations;
-
 import com.amazonaws.services.lambda.runtime.Context;
 import com.amazonaws.services.lambda.runtime.RequestHandler;
 import com.amazonaws.services.lambda.runtime.events.APIGatewayProxyRequestEvent;
@@ -55,24 +37,45 @@
 import com.amazonaws.services.lambda.runtime.events.KinesisFirehoseEvent;
 import com.amazonaws.services.lambda.runtime.events.RabbitMQEvent;
 import com.amazonaws.services.lambda.runtime.events.SNSEvent;
+import com.amazonaws.services.lambda.runtime.events.SQSBatchResponse;
 import com.amazonaws.services.lambda.runtime.events.SQSEvent;
 import com.amazonaws.services.lambda.runtime.events.ScheduledEvent;
-import com.amazonaws.services.lambda.runtime.serialization.PojoSerializer;
-import com.amazonaws.services.lambda.runtime.serialization.events.LambdaEventSerializers;
+import com.amazonaws.services.lambda.runtime.events.StreamsEventResponse;
+import com.amazonaws.services.lambda.runtime.tests.annotations.Event;
 import com.networknt.schema.SpecVersion;
-
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.aspectj.lang.ProceedingJoinPoint;
+import org.aspectj.lang.Signature;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.ArgumentsSource;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
 import software.amazon.lambda.powertools.validation.Validation;
 import software.amazon.lambda.powertools.validation.ValidationConfig;
 import software.amazon.lambda.powertools.validation.ValidationException;
 import software.amazon.lambda.powertools.validation.handlers.GenericSchemaV7APIGatewayProxyRequestEventHandler;
 import software.amazon.lambda.powertools.validation.handlers.GenericSchemaV7StringHandler;
+import software.amazon.lambda.powertools.validation.handlers.KinesisHandlerWithError;
+import software.amazon.lambda.powertools.validation.handlers.SQSHandlerWithError;
 import software.amazon.lambda.powertools.validation.handlers.SQSWithCustomEnvelopeHandler;
 import software.amazon.lambda.powertools.validation.handlers.SQSWithWrongEnvelopeHandler;
+import software.amazon.lambda.powertools.validation.handlers.StandardKinesisHandler;
+import software.amazon.lambda.powertools.validation.handlers.StandardSQSHandler;
 import software.amazon.lambda.powertools.validation.handlers.ValidationInboundAPIGatewayV2HTTPEventHandler;
 import software.amazon.lambda.powertools.validation.model.MyCustomEvent;
 
 
-public class ValidationAspectTest {
+class ValidationAspectTest {
 
     @Mock
     Validation validation;
@@ -167,7 +170,7 @@ void testValidateOutboundJsonSchemaWithHandledExceptions(Object object) throws T
     }
 
     @Test
-    public void testValidateOutboundJsonSchema_APIGWV2() throws Throwable {
+    void testValidateOutboundJsonSchema_APIGWV2() throws Throwable {
         when(validation.schemaVersion()).thenReturn(SpecVersion.VersionFlag.V7);
         when(pjp.getSignature()).thenReturn(signature);
         when(pjp.getSignature().getDeclaringType()).thenReturn(RequestHandler.class);
@@ -187,7 +190,7 @@ public void testValidateOutboundJsonSchema_APIGWV2() throws Throwable {
     }
 
     @Test
-    public void validate_inputOK_schemaInClasspath_shouldValidate() {
+    void validate_inputOK_schemaInClasspath_shouldValidate() {
     	GenericSchemaV7APIGatewayProxyRequestEventHandler handler = new GenericSchemaV7APIGatewayProxyRequestEventHandler();
         APIGatewayProxyRequestEvent event = new APIGatewayProxyRequestEvent();
         event.setBody("{" +
@@ -204,7 +207,7 @@ public void validate_inputOK_schemaInClasspath_shouldValidate() {
     }
 
     @Test
-    public void validate_inputKO_schemaInClasspath_shouldThrowValidationException() {
+    void validate_inputKO_schemaInClasspath_shouldThrowValidationException() {
     	GenericSchemaV7APIGatewayProxyRequestEventHandler handler = new GenericSchemaV7APIGatewayProxyRequestEventHandler();
 
         Map<String, String> headers = new HashMap<>();
@@ -232,7 +235,7 @@ public void validate_inputKO_schemaInClasspath_shouldThrowValidationException()
     }
 
     @Test
-    public void validate_inputOK_schemaInString_shouldValidate() {
+    void validate_inputOK_schemaInString_shouldValidate() {
     	ValidationInboundAPIGatewayV2HTTPEventHandler handler = new ValidationInboundAPIGatewayV2HTTPEventHandler();
         APIGatewayV2HTTPEvent event = new APIGatewayV2HTTPEvent();
         event.setBody("{" +
@@ -248,7 +251,7 @@ public void validate_inputOK_schemaInString_shouldValidate() {
 
 
     @Test
-    public void validate_inputKO_schemaInString_shouldThrowValidationException() {
+    void validate_inputKO_schemaInString_shouldThrowValidationException() {
     	ValidationInboundAPIGatewayV2HTTPEventHandler handler = new ValidationInboundAPIGatewayV2HTTPEventHandler();
 
         Map<String, String> headers = new HashMap<>();
@@ -268,49 +271,77 @@ public void validate_inputKO_schemaInString_shouldThrowValidationException() {
         assertThat(response.getMultiValueHeaders()).isEmpty();
     }
 
-    @Test
-    public void validate_SQS() {
-        PojoSerializer<SQSEvent> pojoSerializer =
-                LambdaEventSerializers.serializerFor(SQSEvent.class, ClassLoader.getSystemClassLoader());
-        SQSEvent event = pojoSerializer.fromJson(this.getClass().getResourceAsStream("/sqs.json"));
-
+    @ParameterizedTest
+    @Event(value = "sqs.json", type = SQSEvent.class)
+    void validate_SQS(SQSEvent event) {
         GenericSchemaV7StringHandler<Object> handler = new GenericSchemaV7StringHandler<>();
         assertThat(handler.handleRequest(event, context)).isEqualTo("OK");
     }
 
-    @Test
-    public void validate_SQS_CustomEnvelopeTakePrecedence() {
-        PojoSerializer<SQSEvent> pojoSerializer =
-                LambdaEventSerializers.serializerFor(SQSEvent.class, ClassLoader.getSystemClassLoader());
-        SQSEvent event = pojoSerializer.fromJson(this.getClass().getResourceAsStream("/sqs_message.json"));
+    @ParameterizedTest
+    @Event(value = "sqs_invalid_messages.json", type = SQSEvent.class)
+    void validate_SQS_with_validation_partial_failure(SQSEvent event) {
+        StandardSQSHandler handler = new StandardSQSHandler();
+        SQSBatchResponse response = handler.handleRequest(event, context);
+        assertThat(response.getBatchItemFailures()).hasSize(2);
+        assertThat(response.getBatchItemFailures().stream().map(SQSBatchResponse.BatchItemFailure::getItemIdentifier).collect(
+                Collectors.toList())).contains("d9144555-9a4f-4ec3-99a0-fc4e625a8db3", "d9144555-9a4f-4ec3-99a0-fc4e625a8db5");
+    }
+
+    @ParameterizedTest
+    @Event(value = "sqs_invalid_messages.json", type = SQSEvent.class)
+    void validate_SQS_with_partial_failure(SQSEvent event) {
+        SQSHandlerWithError handler = new SQSHandlerWithError();
+        SQSBatchResponse response = handler.handleRequest(event, context);
+        assertThat(response.getBatchItemFailures()).hasSize(3);
+        assertThat(response.getBatchItemFailures().stream().map(SQSBatchResponse.BatchItemFailure::getItemIdentifier).collect(
+                Collectors.toList())).contains("d9144555-9a4f-4ec3-99a0-fc4e625a8db3", "d9144555-9a4f-4ec3-99a0-fc4e625a8db5", "1234");
+    }
 
+    @ParameterizedTest
+    @Event(value = "sqs_message.json", type = SQSEvent.class)
+    void validate_SQS_CustomEnvelopeTakePrecedence(SQSEvent event) {
         SQSWithCustomEnvelopeHandler handler = new SQSWithCustomEnvelopeHandler();
         assertThat(handler.handleRequest(event, context)).isEqualTo("OK");
     }
 
-    @Test
-    public void validate_SQS_WrongEnvelope_shouldThrowValidationException() {
-        PojoSerializer<SQSEvent> pojoSerializer =
-                LambdaEventSerializers.serializerFor(SQSEvent.class, ClassLoader.getSystemClassLoader());
-        SQSEvent event = pojoSerializer.fromJson(this.getClass().getResourceAsStream("/sqs_message.json"));
-
+    @ParameterizedTest
+    @Event(value = "sqs_message.json", type = SQSEvent.class)
+    void validate_SQS_WrongEnvelope_shouldThrowValidationException(SQSEvent event) {
         SQSWithWrongEnvelopeHandler handler = new SQSWithWrongEnvelopeHandler();
         assertThatExceptionOfType(ValidationException.class).isThrownBy(() -> handler.handleRequest(event, context));
     }
 
-    @Test
-    public void validate_Kinesis() {
-        PojoSerializer<KinesisEvent> pojoSerializer =
-                LambdaEventSerializers.serializerFor(KinesisEvent.class, ClassLoader.getSystemClassLoader());
-        KinesisEvent event = pojoSerializer.fromJson(this.getClass().getResourceAsStream("/kinesis.json"));
-
+    @ParameterizedTest
+    @Event(value = "kinesis.json", type = KinesisEvent.class)
+    void validate_Kinesis(KinesisEvent event) {
         GenericSchemaV7StringHandler<Object> handler = new GenericSchemaV7StringHandler<>();
         assertThat(handler.handleRequest(event, context)).isEqualTo("OK");
     }
 
+    @ParameterizedTest
+    @Event(value = "kinesis_invalid_messages.json", type = KinesisEvent.class)
+    void validate_Kinesis_with_validation_partial_failure(KinesisEvent event) {
+        StandardKinesisHandler handler = new StandardKinesisHandler();
+        StreamsEventResponse response = handler.handleRequest(event, context);
+        assertThat(response.getBatchItemFailures()).hasSize(2);
+        assertThat(response.getBatchItemFailures().stream().map(StreamsEventResponse.BatchItemFailure::getItemIdentifier).collect(
+                Collectors.toList())).contains("49545115243490985018280067714973144582180062593244200962", "49545115243490985018280067714973144582180062593244200964");
+    }
+
+    @ParameterizedTest
+    @Event(value = "kinesis_invalid_messages.json", type = KinesisEvent.class)
+    void validate_Kinesis_with_partial_failure(KinesisEvent event) {
+        KinesisHandlerWithError handler = new KinesisHandlerWithError();
+        StreamsEventResponse response = handler.handleRequest(event, context);
+        assertThat(response.getBatchItemFailures()).hasSize(3);
+        assertThat(response.getBatchItemFailures().stream().map(StreamsEventResponse.BatchItemFailure::getItemIdentifier).collect(
+                Collectors.toList())).contains("49545115243490985018280067714973144582180062593244200962", "49545115243490985018280067714973144582180062593244200964", "1234");
+    }
+
     @ParameterizedTest
     @MethodSource("provideEventAndEventType")
-    public void validateEEvent(String jsonResource, Class eventClass) throws IOException {
+    void validateEEvent(String jsonResource, Class eventClass) throws IOException {
         Object event = ValidationConfig.get().getObjectMapper()
                 .readValue(this.getClass().getResourceAsStream(jsonResource), eventClass);