When falling back to Lambda default, handle conversion between InputStream and String.

Author: phipag

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

@@ -12,6 +12,7 @@
  */
 package software.amazon.lambda.powertools.kafka.serializers;
 
+import java.io.IOException;
 import java.io.InputStream;
 import java.lang.reflect.Type;
 
@@ -33,6 +34,15 @@ public <T> T fromJson(InputStream input, Type type) {
             return (T) input;
         }
 
+        // If the target type is String, read the input stream as a String
+        if (type.equals(String.class)) {
+            try {
+                return (T) new String(input.readAllBytes());
+            } catch (IOException e) {
+                throw new RuntimeException("Failed to read input stream as String", e);
+            }
+        }
+
         return (T) JacksonFactory.getInstance().getSerializer(type).fromJson(input);
     }
 
@@ -44,6 +54,11 @@ public <T> T fromJson(String input, Type type) {
             return (T) input;
         }
 
+        // If the target type is InputStream, read the input stream as a String
+        if (type.equals(InputStream.class)) {
+            return (T) new String(input).getBytes();
+        }
+
         return (T) JacksonFactory.getInstance().getSerializer(type).fromJson(input);
     }
 }

powertools-kafka/src/test/java/software/amazon/lambda/powertools/kafka/PowertoolsSerializerTest.java

@@ -13,6 +13,7 @@
 package software.amazon.lambda.powertools.kafka;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static software.amazon.lambda.powertools.kafka.testutils.TestUtils.createConsumerRecordsType;
 import static software.amazon.lambda.powertools.kafka.testutils.TestUtils.serializeAvro;
 
@@ -40,6 +41,7 @@
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 
+import software.amazon.lambda.powertools.kafka.serializers.LambdaDefaultDeserializer;
 import software.amazon.lambda.powertools.kafka.serializers.PowertoolsDeserializer;
 import software.amazon.lambda.powertools.kafka.testutils.TestProductPojo;
 
@@ -142,6 +144,62 @@ void shouldHandleInputStreamType() throws IOException {
         assertThat(resultString).isEqualTo(testInput);
     }
 
+    @Test
+    void shouldConvertInputStreamToString() {
+        // When
+        LambdaDefaultDeserializer deserializer = new LambdaDefaultDeserializer();
+
+        // Then
+        String expected = "This is a test string";
+        ByteArrayInputStream inputStream = new ByteArrayInputStream(expected.getBytes());
+
+        // Convert InputStream to String
+        String result = deserializer.fromJson(inputStream, String.class);
+
+        // Verify the result
+        assertThat(result).isEqualTo(expected);
+    }
+
+    @Test
+    void shouldThrowRuntimeExceptionWhenInputStreamIsInvalid() {
+        // When
+        LambdaDefaultDeserializer deserializer = new LambdaDefaultDeserializer();
+
+        // Create a problematic InputStream that throws IOException when read
+        InputStream problematicStream = new InputStream() {
+            @Override
+            public int read() throws IOException {
+                throw new IOException("Simulated IO error");
+            }
+
+            @Override
+            public byte[] readAllBytes() throws IOException {
+                throw new IOException("Simulated IO error");
+            }
+        };
+
+        // Then
+        assertThatThrownBy(() -> deserializer.fromJson(problematicStream, String.class))
+                .isInstanceOf(RuntimeException.class)
+                .hasMessageContaining("Failed to read input stream as String");
+    }
+
+    @Test
+    void shouldConvertStringToByteArray() {
+        // When
+        LambdaDefaultDeserializer deserializer = new LambdaDefaultDeserializer();
+
+        // Then
+        String input = "This is a test string";
+
+        // Convert String to InputStream
+        byte[] result = deserializer.fromJson(input, InputStream.class);
+
+        // Verify the result
+        String resultString = new String(result);
+        assertThat(resultString).isEqualTo(input);
+    }
+
     @ParameterizedTest
     @MethodSource("inputTypes")
     @SetEnvironmentVariable(key = "_HANDLER", value = "software.amazon.lambda.powertools.kafka.testutils.JsonHandler::handleRequest")