DGS-21988 Fix transformation of nullable JSON props (#2516)

* Fix transformation of nullable JSON props

* Minor optimization

Author: rayokota

src/Confluent.SchemaRegistry.Serdes.Json/JsonUtils.cs

@@ -46,7 +46,34 @@ public static async Task<object> Transform(RuleContext ctx, JsonSchema schema, s
             {
                 fieldContext.Type = GetType(schema);
             }
-            
+
+            if (HasMultipleFlags(schema.Type))
+            {
+                JToken jsonObject = JToken.FromObject(message);
+                foreach (JsonObjectType flag in Enum.GetValues(typeof(JsonObjectType)))
+                {
+                    if (schema.Type.HasFlag(flag) && !flag.Equals(default(JsonObjectType)))
+                    {
+                        JsonObjectType originalType = schema.Type;
+                        try
+                        {
+                            schema.Type = flag;
+                            var validator = new JsonSchemaValidator();
+                            var errors = validator.Validate(jsonObject, schema);
+                            if (errors.Count == 0)
+                            {
+                                return await Transform(ctx, schema, path, message,
+                                    fieldTransform).ConfigureAwait(false);
+                            }
+                        }
+                        finally
+                        {
+                            schema.Type = originalType;
+                        }
+                    }
+                }
+
+            }
             if (schema.AllOf.Count > 0 || schema.AnyOf.Count > 0 || schema.OneOf.Count > 0)
             {
                 JToken jsonObject = JToken.FromObject(message);
@@ -139,9 +166,20 @@ public static async Task<object> Transform(RuleContext ctx, JsonSchema schema, s
             }
         }
 
+        private static bool HasMultipleFlags<T>(T flags) where T : Enum
+        {
+            var value = Convert.ToInt32(flags);
+            return value != 0 && (value & (value - 1)) != 0;
+        }
+
         private static RuleContext.Type GetType(JsonSchema schema)
         {
-            switch (schema.Type)
+            return GetType(schema.Type);
+        }
+
+        private static RuleContext.Type GetType(JsonObjectType type)
+        {
+            switch (type)
             {
                 case JsonObjectType.Object:
                     return RuleContext.Type.Record;

test/Confluent.SchemaRegistry.Serdes.UnitTests/JsonSerializeDeserialize.cs

@@ -706,6 +706,57 @@ public void CELFieldTransform()
             Assert.Equal(user.FavoriteNumber, result.FavoriteNumber);
         }
 
+        [Fact]
+        public void CELFieldTransformWithNullable()
+        {
+            var schemaStr = @"{
+              ""type"": ""object"",
+              ""properties"": {
+                ""favorite_color"": {
+                  ""type"": ""string""
+                },
+                ""favorite_number"": {
+                  ""type"": ""number""
+                },
+                ""name"": {
+                  ""type"": [""string"", ""null""]
+                }
+              }
+            }";
+            var schema = new RegisteredSchema("topic-value", 1, 1, schemaStr, SchemaType.Json, null);
+            schema.RuleSet = new RuleSet(new List<Rule>(),
+                new List<Rule>
+                {
+                    new Rule("testCEL", RuleKind.Transform, RuleMode.Write, "CEL_FIELD", null, null,
+                        "typeName == 'STRING' ; value + '-suffix'", null, null, false)
+                }
+            );
+            store[schemaStr] = 1;
+            subjectStore["topic-value"] = new List<RegisteredSchema> { schema };
+            var config = new JsonSerializerConfig
+            {
+                AutoRegisterSchemas = false,
+                UseLatestVersion = true
+            };
+            var serializer = new JsonSerializer<Customer>(schemaRegistryClient, config);
+            var deserializer = new JsonDeserializer<Customer>(schemaRegistryClient);
+
+            var user = new Customer
+            {
+                FavoriteColor = "blue",
+                FavoriteNumber = 100,
+                Name = "awesome"
+            };
+
+            Headers headers = new Headers();
+            var bytes = serializer.SerializeAsync(user, new SerializationContext(MessageComponentType.Value, testTopic, headers)).Result;
+            var result = deserializer.DeserializeAsync(bytes, false, new SerializationContext(MessageComponentType.Value, testTopic, headers)).Result;
+
+            Assert.Equal("awesome-suffix", result.Name);
+            Assert.Equal("blue-suffix", result.FavoriteColor);
+            Assert.Equal(user.FavoriteNumber, result.FavoriteNumber);
+        }
+
         [Fact]
         public void CELFieldTransformWithDef()
         {