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

* DGS-22077 Handle evolution during field transformation

* Minor fix

* Minor cleanup

* Add new header

Author: rayokota

src/confluent_kafka/schema_registry/_async/schema_registry_client.py

@@ -438,9 +438,10 @@ async def send_request(
                              " application/json"}
 
         if body is not None:
-            body = json.dumps(body)
-            headers = {'Content-Length': str(len(body)),
-                       'Content-Type': "application/vnd.schemaregistry.v1+json"}
+            body_str = json.dumps(body)
+            headers = {'Content-Length': str(len(body_str)),
+                       'Content-Type': "application/vnd.schemaregistry.v1+json",
+                       'Confluent-Accept-Unknown-Properties': "true"}
 
         if self.bearer_auth_credentials_source:
             await self.handle_bearer_auth(headers)

src/confluent_kafka/schema_registry/_sync/schema_registry_client.py

@@ -438,9 +438,10 @@ def send_request(
                              " application/json"}
 
         if body is not None:
-            body = json.dumps(body)
-            headers = {'Content-Length': str(len(body)),
-                       'Content-Type': "application/vnd.schemaregistry.v1+json"}
+            body_str = json.dumps(body)
+            headers = {'Content-Length': str(len(body_str)),
+                       'Content-Type': "application/vnd.schemaregistry.v1+json",
+                       'Confluent-Accept-Unknown-Properties': "true"}
 
         if self.bearer_auth_credentials_source:
             self.handle_bearer_auth(headers)

src/confluent_kafka/schema_registry/common/avro.py

@@ -121,6 +121,8 @@ def transform(
         elif schema_type == 'record':
             fields = schema["fields"]
             for field in fields:
+                if field["name"] not in message:
+                    continue
                 _transform_field(ctx, schema, field, message, field_transform)
             return message
 

tests/schema_registry/_async/test_avro_serdes.py

@@ -721,6 +721,68 @@ async def test_avro_cel_field_transform():
     assert obj2 == newobj
 
 
+async def test_avro_cel_field_transform_missing_prop():
+    conf = {'url': _BASE_URL}
+    client = AsyncSchemaRegistryClient.new_client(conf)
+    ser_conf = {'auto.register.schemas': False, 'use.latest.version': True}
+    schema = {
+        'type': 'record',
+        'name': 'test',
+        'fields': [
+            {'name': 'intField', 'type': 'int'},
+            {'name': 'doubleField', 'type': 'double'},
+            {'name': 'stringField', 'type': 'string'},
+            {'name': 'booleanField', 'type': 'boolean'},
+            {'name': 'bytesField', 'type': 'bytes'},
+            {'name': 'missing', 'type': ['null', 'string'], 'default': None},
+        ]
+    }
+
+    rule = Rule(
+        "test-cel",
+        "",
+        RuleKind.TRANSFORM,
+        RuleMode.WRITEREAD,
+        "CEL_FIELD",
+        None,
+        None,
+        "name == 'stringField' ; value + '-suffix'",
+        None,
+        None,
+        False
+    )
+    await client.register_schema(_SUBJECT, Schema(
+        json.dumps(schema),
+        "AVRO",
+        [],
+        None,
+        RuleSet(None, [rule])
+    ))
+
+    obj = {
+        'intField': 123,
+        'doubleField': 45.67,
+        'stringField': 'hi',
+        'booleanField': True,
+        'bytesField': b'foobar',
+    }
+    ser = await AsyncAvroSerializer(client, schema_str=None, conf=ser_conf)
+    ser_ctx = SerializationContext(_TOPIC, MessageField.VALUE)
+    obj_bytes = await ser(obj, ser_ctx)
+
+    obj2 = {
+        'intField': 123,
+        'doubleField': 45.67,
+        'stringField': 'hi-suffix-suffix',
+        'booleanField': True,
+        'bytesField': b'foobar',
+        'missing': None,
+    }
+    deser = await AsyncAvroDeserializer(client)
+    newobj = await deser(obj_bytes, ser_ctx)
+    assert obj2 == newobj
+
+
 async def test_avro_cel_field_transform_disable():
     conf = {'url': _BASE_URL}
     client = AsyncSchemaRegistryClient.new_client(conf)

tests/schema_registry/_sync/test_avro_serdes.py

@@ -721,6 +721,68 @@ def test_avro_cel_field_transform():
     assert obj2 == newobj
 
 
+def test_avro_cel_field_transform_missing_prop():
+    conf = {'url': _BASE_URL}
+    client = SchemaRegistryClient.new_client(conf)
+    ser_conf = {'auto.register.schemas': False, 'use.latest.version': True}
+    schema = {
+        'type': 'record',
+        'name': 'test',
+        'fields': [
+            {'name': 'intField', 'type': 'int'},
+            {'name': 'doubleField', 'type': 'double'},
+            {'name': 'stringField', 'type': 'string'},
+            {'name': 'booleanField', 'type': 'boolean'},
+            {'name': 'bytesField', 'type': 'bytes'},
+            {'name': 'missing', 'type': ['null', 'string'], 'default': None},
+        ]
+    }
+
+    rule = Rule(
+        "test-cel",
+        "",
+        RuleKind.TRANSFORM,
+        RuleMode.WRITEREAD,
+        "CEL_FIELD",
+        None,
+        None,
+        "name == 'stringField' ; value + '-suffix'",
+        None,
+        None,
+        False
+    )
+    client.register_schema(_SUBJECT, Schema(
+        json.dumps(schema),
+        "AVRO",
+        [],
+        None,
+        RuleSet(None, [rule])
+    ))
+
+    obj = {
+        'intField': 123,
+        'doubleField': 45.67,
+        'stringField': 'hi',
+        'booleanField': True,
+        'bytesField': b'foobar',
+    }
+    ser = AvroSerializer(client, schema_str=None, conf=ser_conf)
+    ser_ctx = SerializationContext(_TOPIC, MessageField.VALUE)
+    obj_bytes = ser(obj, ser_ctx)
+
+    obj2 = {
+        'intField': 123,
+        'doubleField': 45.67,
+        'stringField': 'hi-suffix-suffix',
+        'booleanField': True,
+        'bytesField': b'foobar',
+        'missing': None,
+    }
+    deser = AvroDeserializer(client)
+    newobj = deser(obj_bytes, ser_ctx)
+    assert obj2 == newobj
+
+
 def test_avro_cel_field_transform_disable():
     conf = {'url': _BASE_URL}
     client = SchemaRegistryClient.new_client(conf)