feat: improve error handling in serializers and add tests for corrupted data

hjgraca · hjgraca · commit c64eccf5ffd1 · 2025-06-20T18:39:40.000+01:00
diff --git a/libraries/src/AWS.Lambda.Powertools.Kafka.Avro/PowertoolsKafkaAvroSerializer.cs b/libraries/src/AWS.Lambda.Powertools.Kafka.Avro/PowertoolsKafkaAvroSerializer.cs
@@ -93,18 +93,10 @@ public PowertoolsKafkaAvroSerializer(JsonSerializerContext serializerContext) :
                 "Consider using an alternative Deserializer.");
         }
 
-        try
-        {
-            using var stream = new MemoryStream(data);
-            var decoder = new BinaryDecoder(stream);
-            var reader = new SpecificDatumReader<object>(schema, schema);
-            return reader.Read(null!, decoder);
-        }
-        catch (Exception ex)
-        {
-            throw new System.Runtime.Serialization.SerializationException(
-                $"Failed to deserialize {(isKey ? "key" : "value")} data: {ex.Message}", ex);
-        }
+        using var stream = new MemoryStream(data);
+        var decoder = new BinaryDecoder(stream);
+        var reader = new SpecificDatumReader<object>(schema, schema);
+        return reader.Read(null!, decoder);
     }
 
     /// <summary>
diff --git a/libraries/src/AWS.Lambda.Powertools.Kafka.Protobuf/PowertoolsKafkaProtobufSerializer.cs b/libraries/src/AWS.Lambda.Powertools.Kafka.Protobuf/PowertoolsKafkaProtobufSerializer.cs
@@ -101,15 +101,7 @@ public PowertoolsKafkaProtobufSerializer(JsonSerializerContext serializerContext
             throw new InvalidOperationException($"Could not find Protobuf parser for type {targetType.Name}");
         }
 
-        try
-        {
-            return DeserializeByStrategy(data, parser, schemaMetadata);
-        }
-        catch (Exception ex)
-        {
-            throw new System.Runtime.Serialization.SerializationException(
-                $"Failed to deserialize {(isKey ? "key" : "value")} data: {ex.Message}", ex);
-        }
+        return DeserializeByStrategy(data, parser, schemaMetadata);
     }
 
     /// <summary>
@@ -156,6 +148,15 @@ private IMessage DeserializeConfluentFormat(byte[] data, MessageParser parser)
         using var inputStream = new MemoryStream(data);
         using var codedInput = new CodedInputStream(inputStream);
 
+        /*
+            ReadSInt32() behavior:
+               ReadSInt32() properly handles signed varint encoding using ZigZag encoding
+               ZigZag encoding maps signed integers to unsigned integers: (n << 1) ^ (n >> 31)
+               This allows both positive and negative numbers to be efficiently encoded
+               The key insight is that Confluent Schema Registry uses signed varint encoding for the message index count, not unsigned length encoding.
+               The ByteUtils.readVarint() in Java typically reads signed varints, which corresponds to ReadSInt32() in C# Google.Protobuf.
+         */
+        
         // Read number of message indexes
         var indexCount = codedInput.ReadSInt32();
         
diff --git a/libraries/src/AWS.Lambda.Powertools.Kafka/PowertoolsKafkaSerializerBase.cs b/libraries/src/AWS.Lambda.Powertools.Kafka/PowertoolsKafkaSerializerBase.cs
@@ -328,7 +328,7 @@ private void ProcessValue(JsonElement recordElement, object record, Type recordT
                 }
                 catch (Exception ex)
                 {
-                    throw new SerializationException(ex.Message);
+                    throw new SerializationException($"Failed to deserialize value data: {ex.Message}", ex);
                 }
             }
         }
diff --git a/libraries/tests/AWS.Lambda.Powertools.Kafka.Tests/ErrorHandlingTests.cs b/libraries/tests/AWS.Lambda.Powertools.Kafka.Tests/ErrorHandlingTests.cs
@@ -0,0 +1,91 @@
+/*
+ * Copyright JsonCons.Net authors. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file 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.
+ */
+
+using System.Runtime.Serialization;
+using System.Text;
+using AWS.Lambda.Powertools.Kafka.Avro;
+using AWS.Lambda.Powertools.Kafka.Json;
+using AWS.Lambda.Powertools.Kafka.Protobuf;
+
+namespace AWS.Lambda.Powertools.Kafka.Tests;
+
+public class ErrorHandlingTests
+{
+    [Theory]
+    [InlineData(typeof(PowertoolsKafkaJsonSerializer))]
+    [InlineData(typeof(PowertoolsKafkaAvroSerializer))]
+    [InlineData(typeof(PowertoolsKafkaProtobufSerializer))]
+    public void AllSerializers_WithCorruptedKeyData_ThrowSerializationException(Type serializerType)
+    {
+        // Arrange
+        var serializer = (PowertoolsKafkaSerializerBase)Activator.CreateInstance(serializerType)!;
+        var corruptedData = new byte[] { 0xDE, 0xAD, 0xBE, 0xEF };
+        
+        string kafkaEventJson = CreateKafkaEvent(
+            Convert.ToBase64String(corruptedData),
+            Convert.ToBase64String(Encoding.UTF8.GetBytes("valid-value"))
+        );
+        
+        using var stream = new MemoryStream(Encoding.UTF8.GetBytes(kafkaEventJson));
+
+        // Act & Assert
+        var ex = Assert.Throws<SerializationException>(() =>
+            serializer.Deserialize<ConsumerRecords<TestModel, string>>(stream));
+        
+        Assert.Contains("Failed to deserialize key data", ex.Message);
+    }
+
+    [Theory]
+    [InlineData(typeof(PowertoolsKafkaJsonSerializer))]
+    [InlineData(typeof(PowertoolsKafkaAvroSerializer))]
+    [InlineData(typeof(PowertoolsKafkaProtobufSerializer))]
+    public void AllSerializers_WithCorruptedValueData_ThrowSerializationException(Type serializerType)
+    {
+        // Arrange
+        var serializer = (PowertoolsKafkaSerializerBase)Activator.CreateInstance(serializerType)!;
+        var corruptedData = new byte[] { 0xDE, 0xAD, 0xBE, 0xEF };
+        
+        string kafkaEventJson = CreateKafkaEvent(
+            Convert.ToBase64String(Encoding.UTF8.GetBytes("valid-key")),
+            Convert.ToBase64String(corruptedData)
+        );
+        
+        using var stream = new MemoryStream(Encoding.UTF8.GetBytes(kafkaEventJson));
+
+        // Act & Assert
+        var ex = Assert.Throws<SerializationException>(() =>
+            serializer.Deserialize<ConsumerRecords<string, TestModel>>(stream));
+        
+        Assert.Contains("Failed to deserialize value data", ex.Message);
+    }
+
+    private string CreateKafkaEvent(string keyValue, string valueValue)
+    {
+        return @$"{{
+            ""eventSource"": ""aws:kafka"",
+            ""records"": {{
+                ""mytopic-0"": [
+                    {{
+                        ""topic"": ""mytopic"",
+                        ""partition"": 0,
+                        ""offset"": 15,
+                        ""key"": ""{keyValue}"",
+                        ""value"": ""{valueValue}""
+                    }}
+                ]
+            }}
+        }}";
+    }
+}
diff --git a/libraries/tests/AWS.Lambda.Powertools.Kafka.Tests/Json/PowertoolsKafkaJsonSerializerTests.cs b/libraries/tests/AWS.Lambda.Powertools.Kafka.Tests/Json/PowertoolsKafkaJsonSerializerTests.cs
@@ -13,6 +13,7 @@
  * permissions and limitations under the License.
  */
 
+using System.Runtime.Serialization;
 using System.Text;
 using System.Text.Json;
 using System.Text.Json.Serialization;
@@ -27,47 +28,22 @@ public void Deserialize_KafkaEventWithJsonPayload_DeserializesToCorrectType()
     {
         // Arrange
         var serializer = new PowertoolsKafkaJsonSerializer();
-        string kafkaEventJson = File.ReadAllText("Json/kafka-json-event.json");
+        var testModel = new TestModel { Name = "Test Product", Value = 123 };
+        var jsonValue = JsonSerializer.Serialize(testModel);
+        var base64Value = Convert.ToBase64String(Encoding.UTF8.GetBytes(jsonValue));
+        
+        string kafkaEventJson = CreateKafkaEvent("NDI=", base64Value); // Key is 42 in base64
         using var stream = new MemoryStream(Encoding.UTF8.GetBytes(kafkaEventJson));
 
         // Act
-        var result = serializer.Deserialize<ConsumerRecords<string, JsonProduct>>(stream);
+        var result = serializer.Deserialize<ConsumerRecords<int, TestModel>>(stream);
 
         // Assert
         Assert.NotNull(result);
-        Assert.Equal("aws:kafka", result.EventSource);
-
-        // Verify records were deserialized
-        Assert.True(result.Records.ContainsKey("mytopic-0"));
-        var records = result.Records["mytopic-0"];
-        Assert.Equal(3, records.Count);
-
-        // Verify first record's content
-        var firstRecord = records[0];
-        Assert.Equal("mytopic", firstRecord.Topic);
-        Assert.Equal(0, firstRecord.Partition);
-        Assert.Equal(15, firstRecord.Offset);
-        Assert.Equal("recordKey", firstRecord.Key);
-
-        // Verify deserialized JSON value
-        var product = firstRecord.Value;
-        Assert.Equal("product5", product.Name);
-        Assert.Equal(12345, product.Id);
-        Assert.Equal(45, product.Price);
-
-        // Verify second record
-        var secondRecord = records[1];
-        var p2 = secondRecord.Value;
-        Assert.Equal("product5", p2.Name);
-        Assert.Equal(12345, p2.Id);
-        Assert.Equal(45, p2.Price);
-
-        // Verify third record
-        var thirdRecord = records[2];
-        var p3 = thirdRecord.Value;
-        Assert.Equal("product5", p3.Name);
-        Assert.Equal(12345, p3.Id);
-        Assert.Equal(45, p3.Price);
+        var record = result.First();
+        Assert.Equal(42, record.Key);
+        Assert.Equal("Test Product", record.Value.Name);
+        Assert.Equal(123, record.Value.Value);
     }
 
     [Fact]
diff --git a/libraries/tests/AWS.Lambda.Powertools.Kafka.Tests/PowertoolsKafkaSerializerBaseTests.cs b/libraries/tests/AWS.Lambda.Powertools.Kafka.Tests/PowertoolsKafkaSerializerBaseTests.cs
@@ -46,55 +46,51 @@ public TestKafkaSerializer(JsonSerializerOptions options, JsonSerializerContext
                 : base(options, context)
             {
             }
-            
+
             // Implementation of the abstract method for test purposes
-            protected override object? DeserializeComplexTypeFormat(byte[] data, 
+            protected override object? DeserializeComplexTypeFormat(byte[] data,
                 Type targetType, bool isKey, SchemaMetadata? schemaMetadata = null)
             {
-                try
+                // Test implementation using JSON for all complex types
+                var jsonStr = Encoding.UTF8.GetString(data);
+
+                if (SerializerContext != null)
                 {
-                    // Test implementation using JSON for all complex types
-                    var jsonStr = Encoding.UTF8.GetString(data);
-                    
-                    if (SerializerContext != null)
+                    var typeInfo = SerializerContext.GetTypeInfo(targetType);
+                    if (typeInfo != null)
                     {
-                        var typeInfo = SerializerContext.GetTypeInfo(targetType);
-                        if (typeInfo != null)
-                        {
-                            return JsonSerializer.Deserialize(jsonStr, typeInfo);
-                        }
+                        return JsonSerializer.Deserialize(jsonStr, typeInfo);
                     }
-                    
-                    return JsonSerializer.Deserialize(jsonStr, targetType, JsonOptions);
-                }
-                catch
-                {
-                    return null;
                 }
+
+                return JsonSerializer.Deserialize(jsonStr, targetType, JsonOptions);
             }
-            
+
             // Expose protected methods for direct testing
-            public object? TestDeserializeFormatSpecific(byte[] data, Type targetType, bool isKey, SchemaMetadata? schemaMetadata = null)
+            public object? TestDeserializeFormatSpecific(byte[] data, Type targetType, bool isKey,
+                SchemaMetadata? schemaMetadata = null)
             {
                 return DeserializeFormatSpecific(data, targetType, isKey, schemaMetadata);
             }
-            
-            public object? TestDeserializeComplexTypeFormat(byte[] data, Type targetType, bool isKey, SchemaMetadata? schemaMetadata = null)
+
+            public object? TestDeserializeComplexTypeFormat(byte[] data, Type targetType, bool isKey,
+                SchemaMetadata? schemaMetadata = null)
             {
                 return DeserializeComplexTypeFormat(data, targetType, isKey, schemaMetadata);
             }
-            
+
             public object? TestDeserializePrimitiveValue(byte[] data, Type targetType)
             {
                 return DeserializePrimitiveValue(data, targetType);
             }
-            
+
             public bool TestIsPrimitiveOrSimpleType(Type type)
             {
                 return IsPrimitiveOrSimpleType(type);
             }
-            
-            public object TestDeserializeValue(string base64Value, Type valueType, SchemaMetadata? schemaMetadata = null)
+
+            public object TestDeserializeValue(string base64Value, Type valueType,
+                SchemaMetadata? schemaMetadata = null)
             {
                 return DeserializeValue(base64Value, valueType, schemaMetadata);
             }
@@ -610,7 +606,9 @@ public void DeserializeFormatSpecific_PrimitiveType_UsesDeserializePrimitiveValu
             var stringBytes = Encoding.UTF8.GetBytes("primitive-test");
 
             // Act
-            var result = serializer.TestDeserializeFormatSpecific(stringBytes, typeof(string), isKey: false, schemaMetadata: null);
+            var result =
+                serializer.TestDeserializeFormatSpecific(stringBytes, typeof(string), isKey: false,
+                    schemaMetadata: null);
 
             // Assert
             Assert.Equal("primitive-test", result);
@@ -625,7 +623,9 @@ public void DeserializeFormatSpecific_ComplexType_UsesDeserializeComplexTypeForm
             var jsonBytes = Encoding.UTF8.GetBytes(JsonSerializer.Serialize(complexObject));
 
             // Act
-            var result = serializer.TestDeserializeFormatSpecific(jsonBytes, typeof(TestModel), isKey: false, schemaMetadata: null);
+            var result =
+                serializer.TestDeserializeFormatSpecific(jsonBytes, typeof(TestModel), isKey: false,
+                    schemaMetadata: null);
 
             // Assert
             Assert.NotNull(result);
@@ -643,7 +643,9 @@ public void DeserializeComplexTypeFormat_ValidJson_DeserializesCorrectly()
             var jsonBytes = Encoding.UTF8.GetBytes(JsonSerializer.Serialize(complexObject));
 
             // Act
-            var result = serializer.TestDeserializeComplexTypeFormat(jsonBytes, typeof(TestModel), isKey: true, schemaMetadata: null);
+            var result =
+                serializer.TestDeserializeComplexTypeFormat(jsonBytes, typeof(TestModel), isKey: true,
+                    schemaMetadata: null);
 
             // Assert
             Assert.NotNull(result);
@@ -653,17 +655,19 @@ public void DeserializeComplexTypeFormat_ValidJson_DeserializesCorrectly()
         }
 
         [Fact]
-        public void DeserializeComplexTypeFormat_InvalidJson_ReturnsNull()
+        public void DeserializeComplexTypeFormat_InvalidJson_ThrowsException()
         {
             // Arrange
             var serializer = new TestKafkaSerializer();
             var invalidBytes = new byte[] { 0xDE, 0xAD, 0xBE, 0xEF }; // Invalid JSON data
 
-            // Act
-            var result = serializer.TestDeserializeComplexTypeFormat(invalidBytes, typeof(TestModel), isKey: true, schemaMetadata: null);
+            // Act & Assert
+            // The TestKafkaSerializer throws JsonException directly for invalid JSON
+            var ex = Assert.Throws<JsonException>(() =>
+                serializer.TestDeserializeComplexTypeFormat(invalidBytes, typeof(TestModel), isKey: true,
+                    schemaMetadata: null));
 
-            // Assert
-            Assert.Null(result);
+            Assert.Contains("invalid", ex.Message.ToLower());
         }
 
         [Fact]
@@ -702,18 +706,18 @@ public void IsPrimitiveOrSimpleType_ChecksVariousTypes()
         {
             // Arrange
             var serializer = new TestKafkaSerializer();
-            
+
             // Act & Assert
             // Primitive types
             Assert.True(serializer.TestIsPrimitiveOrSimpleType(typeof(int)));
             Assert.True(serializer.TestIsPrimitiveOrSimpleType(typeof(long)));
             Assert.True(serializer.TestIsPrimitiveOrSimpleType(typeof(bool)));
-            
+
             // Simple types
             Assert.True(serializer.TestIsPrimitiveOrSimpleType(typeof(string)));
             Assert.True(serializer.TestIsPrimitiveOrSimpleType(typeof(Guid)));
             Assert.True(serializer.TestIsPrimitiveOrSimpleType(typeof(DateTime)));
-            
+
             // Complex types
             Assert.False(serializer.TestIsPrimitiveOrSimpleType(typeof(TestModel)));
             Assert.False(serializer.TestIsPrimitiveOrSimpleType(typeof(Dictionary<string, int>)));
@@ -758,5 +762,4 @@ public class TestModel
         public string Name { get; set; }
         public int Value { get; set; }
     }
-}
-
+}

Original file line number	Diff line number	Diff line change
`@@ -328,7 +328,7 @@ private void ProcessValue(JsonElement recordElement, object record, Type recordT`
`328`	`328`	`}`
`329`	`329`	`catch (Exception ex)`
`330`	`330`	`{`
`331`		`- throw new SerializationException(ex.Message);`
	`331`	`+ throw new SerializationException($"Failed to deserialize value data: {ex.Message}", ex);`
`332`	`332`	`}`
`333`	`333`	`}`
`334`	`334`	`}`