Raise a runtime exception when the KafkaEvent is invalid.

Author: phipag

powertools-kafka/src/main/java/software/amazon/lambda/powertools/kafka/serializers/AbstractKafkaDeserializer.java

@@ -118,8 +118,13 @@ private boolean isConsumerRecordsType(Type type) {
 
     private <K, V> ConsumerRecords<K, V> convertToConsumerRecords(KafkaEvent kafkaEvent, Class<K> keyType,
             Class<V> valueType) {
+        // Validate that this is actually a Kafka event by checking for required properties
+        if (kafkaEvent == null || kafkaEvent.getEventSource() == null) {
+            throw new RuntimeException(
+                    "Failed to deserialize Lambda handler input to ConsumerRecords: Input is not a valid Kafka event.");
+        }
 
-        if (kafkaEvent == null || kafkaEvent.getRecords() == null) {
+        if (kafkaEvent.getRecords() == null) {
             return ConsumerRecords.empty();
         }
 
@@ -138,7 +143,7 @@ private <K, V> ConsumerRecords<K, V> convertToConsumerRecords(KafkaEvent kafkaEv
 
         return createConsumerRecords(recordsMap);
     }
-    
+
     /**
      * Creates ConsumerRecords with compatibility for both Kafka 3.x.x and 4.x.x.
      * 
@@ -147,7 +152,8 @@ private <K, V> ConsumerRecords<K, V> convertToConsumerRecords(KafkaEvent kafkaEv
      * @param records Map of records by topic partition
      * @return ConsumerRecords instance
      */
-    protected <K, V> ConsumerRecords<K, V> createConsumerRecords(Map<TopicPartition, List<ConsumerRecord<K, V>>> records) {
+    protected <K, V> ConsumerRecords<K, V> createConsumerRecords(
+            Map<TopicPartition, List<ConsumerRecord<K, V>>> records) {
         try {
             // Try to use the Kafka 4.x.x constructor with nextOffsets parameter
             return new ConsumerRecords<>(records, Map.of());

powertools-kafka/src/test/java/software/amazon/lambda/powertools/kafka/serializers/AbstractKafkaDeserializerTest.java

@@ -307,6 +307,42 @@ void shouldHandleNullRecords(InputType inputType) {
         assertThat(records.count()).isZero();
     }
 
+    @ParameterizedTest
+    @MethodSource("inputTypes")
+    void shouldThrowExceptionWhenEventSourceIsNull(InputType inputType) {
+        // Given
+        // Create a JSON without eventSource property
+        String kafkaJson = "{\n" +
+                "  \"records\": {\n" +
+                "    \"test-topic-1\": [\n" +
+                "      {\n" +
+                "        \"topic\": \"test-topic-1\",\n" +
+                "        \"partition\": 0,\n" +
+                "        \"offset\": 15,\n" +
+                "        \"timestamp\": 1545084650987,\n" +
+                "        \"timestampType\": \"CREATE_TIME\",\n" +
+                "        \"key\": null,\n" +
+                "        \"value\": null,\n" +
+                "        \"headers\": []\n" +
+                "      }\n" +
+                "    ]\n" +
+                "  }\n" +
+                "}";
+        Type type = TestUtils.createConsumerRecordsType(String.class, TestProductPojo.class);
+
+        // When/Then
+        if (inputType == InputType.INPUT_STREAM) {
+            ByteArrayInputStream inputStream = new ByteArrayInputStream(kafkaJson.getBytes());
+            assertThatThrownBy(() -> deserializer.fromJson(inputStream, type))
+                    .isInstanceOf(RuntimeException.class)
+                    .hasMessageContaining("Input is not a valid Kafka event");
+        } else {
+            assertThatThrownBy(() -> deserializer.fromJson(kafkaJson, type))
+                    .isInstanceOf(RuntimeException.class)
+                    .hasMessageContaining("Input is not a valid Kafka event");
+        }
+    }
+
     static Stream<Arguments> primitiveTypesProvider() {
         return Stream.of(
                 // For each primitive type, test with both INPUT_STREAM and STRING