DGS-22077 Handle evolution during field transformation (#2541)

* DGS-22077 Handle evolution during field transformation

* Minor rename

* Fix test

* Add new header

Author: rayokota

src/Confluent.SchemaRegistry.Serdes.Avro/AvroUtils.cs

@@ -114,23 +114,40 @@ public static async Task<object> Transform(RuleContext ctx, Avro.Schema schema,
                     return await Utils.TransformDictionaryAsync(message, mapTransformer).ConfigureAwait(false);
                 case Avro.Schema.Type.Record:
                     RecordSchema rs = (RecordSchema)schema;
+                    if (message is ISpecificRecord)
+                    {
+                        ISpecificRecord specificRecord = (ISpecificRecord)message;
+                        rs = (RecordSchema)specificRecord.Schema;
+                    }
+                    else if (message is GenericRecord)
+                    {
+                        GenericRecord genericRecord = (GenericRecord)message;
+                        rs = (RecordSchema)genericRecord.Schema;
+                    }
                     foreach (Field f in rs.Fields)
                     {
+                        Field originalField = null;
+                        if (!((RecordSchema)schema).TryGetField(f.Name, out originalField))
+                        {
+                            originalField = f;
+                        }
+
                         string fullName = rs.Fullname + "." + f.Name;
-                        using (ctx.EnterField(message, fullName, f.Name, GetType(f.Schema), GetInlineTags(f)))
+                        using (ctx.EnterField(message, fullName, f.Name, GetType(originalField.Schema), GetInlineTags(originalField)))
                         {
                             if (message is ISpecificRecord)
                             {
                                 ISpecificRecord specificRecord = (ISpecificRecord)message;
                                 object value = specificRecord.Get(f.Pos);
-                                object newValue = await Transform(ctx, f.Schema, value, fieldTransform).ConfigureAwait(false);
+                                object newValue = await Transform(ctx, originalField.Schema, value, fieldTransform).ConfigureAwait(false);
                                 if (ctx.Rule.Kind == RuleKind.Condition)
                                 {
                                     if (newValue is bool b && !b)
                                     {
                                         throw new RuleConditionException(ctx.Rule);
                                     }
-                                } else
+                                }
+                                else
                                 {
                                     specificRecord.Put(f.Pos, newValue);
                                 }
@@ -139,7 +156,7 @@ public static async Task<object> Transform(RuleContext ctx, Avro.Schema schema,
                             {
                                 GenericRecord genericRecord = (GenericRecord)message;
                                 object value = genericRecord.GetValue(f.Pos);
-                                object newValue = await Transform(ctx, f.Schema, value, fieldTransform).ConfigureAwait(false);
+                                object newValue = await Transform(ctx, originalField.Schema, value, fieldTransform).ConfigureAwait(false);
                                 if (ctx.Rule.Kind == RuleKind.Condition)
                                 {
                                     if (newValue is bool b && !b)

src/Confluent.SchemaRegistry/Rest/RestService.cs

@@ -36,8 +36,6 @@ public class RestService : IRestService
 
         private static readonly string acceptHeader = string.Join(", ", Versions.PreferredResponseTypes);
 
-        private static readonly string acceptVersionHeader = "8.0";
-
         public const int DefaultMaxRetries = 3;
 
         public const int DefaultRetriesWaitMs = 1000;
@@ -447,7 +445,7 @@ private async Task<HttpRequestMessage> CreateRequest(string endPoint, HttpMethod
         {
             HttpRequestMessage request = new HttpRequestMessage(method, endPoint);
             request.Headers.Add("Accept", acceptHeader);
-            request.Headers.Add("Accept-Version", acceptVersionHeader);
+            request.Headers.Add("Confluent-Accept-Unknown-Properties", "true");
             if (jsonBody.Length != 0)
             {
                 string stringContent = string.Join("\n", jsonBody.Select(x => JsonConvert.SerializeObject(x)));

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

@@ -399,6 +399,52 @@ public void ISpecificRecordCELFieldTransform()
             Assert.Equal(user.favorite_number, result.favorite_number);
         }
 
+        [Fact]
+        public void ISpecificRecordCELFieldTransformMissingProp()
+        {
+            var schemaStr =
+                "{\"type\":\"record\",\"name\":\"User\",\"namespace\":\"Confluent.Kafka.Examples.AvroSpecific\",\"fields\":[" +
+                "{\"name\":\"name\",\"type\":\"string\"}," +
+                "{\"name\":\"favorite_number\",\"type\":[\"int\",\"null\"]}," +
+                "{\"name\":\"favorite_color\",\"type\":[\"string\",\"null\"]}," +
+                "{\"name\":\"missing\",\"type\":[\"null\",\"string\"],\"default\":null}]}";
+
+            var schema = new RegisteredSchema("topic-value", 1, 1, schemaStr, SchemaType.Avro, null);
+            schema.RuleSet = new RuleSet(new List<Rule>(),
+                new List<Rule>
+                {
+                    new Rule("testCEL", RuleKind.Transform, RuleMode.WriteRead, "CEL_FIELD", null, null,
+                        "typeName == 'STRING' ; value + '-suffix'", null, null, false)
+                }
+            );
+            store[schemaStr] = 1;
+            subjectStore["topic-value"] = new List<RegisteredSchema> { schema };
+            var config = new AvroSerializerConfig
+            {
+                AutoRegisterSchemas = false,
+                UseLatestVersion = true
+            };
+            var serializer = new AvroSerializer<UserWithPic>(schemaRegistryClient, config);
+            var deserializer = new AvroDeserializer<User>(schemaRegistryClient, null);
+
+            var pic = new byte[] { 1, 2, 3 };
+            var user = new UserWithPic()
+            {
+                favorite_color = "blue",
+                favorite_number = 100,
+                name = "awesome",
+                picture = pic
+            };
+
+            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-suffix", result.name);
+            Assert.Equal("blue-suffix-suffix", result.favorite_color);
+            Assert.Equal(user.favorite_number, result.favorite_number);
+        }
+
         [Fact]
         public void ISpecificRecordCELFieldTransformDisable()
         {