avro: fix primitive and union schema parsing bugs

This commit cleans up a few bugs in the _schema_loads function:

We use primitive types retrieved from the Confluent Registry and encountered an
issue where _schema_loads would cause json deserialization errors by double quoting
valid primitive declarations. Previous tests included incorrectly specified primitive
declarations, according to the Avro spec primitive declarations are valid JSON documents,
but they had been specified as strings of their type name with no quoting. I fixed the
tests as well as the issue in _schema_loads

Somewhat separately, there was also an issue with Avro union types. _schema_loads was
incorrectly causing json serialization errors for unions because it included them
on accident with its special-casing of primitive declarations. I added a check for
json arrays to exclude them from the special casing. I also had to add a check
later to ensure the _schema_name property was special-cased to None for unions.
This should have no impact on names in the registry because _schema_name isn't
used at all for the recommended subject name strategy with unions.

Author: dylrich

src/confluent_kafka/schema_registry/avro.py

@@ -62,8 +62,8 @@ def _schema_loads(schema_str):
     schema_str = schema_str.strip()
 
     # canonical form primitive declarations are not supported
-    if schema_str[0] != "{":
-        schema_str = '{"type":"' + schema_str + '"}'
+    if schema_str[0] != "{" and schema_str[0] != "[":
+        schema_str = '{"type":' + schema_str + '}'
 
     return Schema(schema_str, schema_type='AVRO')
 
@@ -192,11 +192,20 @@ def __init__(self, schema_registry_client, schema_str,
         schema = _schema_loads(schema_str)
         schema_dict = loads(schema.schema_str)
         parsed_schema = parse_schema(schema_dict)
-        # The Avro spec states primitives have a name equal to their type
-        # i.e. {"type": "string"} has a name of string.
-        # This function does not comply.
-        # https://github.com/fastavro/fastavro/issues/415
-        schema_name = parsed_schema.get('name', schema_dict['type'])
+
+        if isinstance(parsed_schema, list):
+            # if parsed_schema is a list, we have an Avro union and there
+            # is no valid schema name. This is fine because the only use of
+            # schema_name is for supplying the subject name to the registry
+            # and union types should use topic_subject_name_strategy, which
+            # just discards the schema name anyway
+            schema_name = None
+        else:
+            # The Avro spec states primitives have a name equal to their type
+            # i.e. {"type": "string"} has a name of string.
+            # This function does not comply.
+            # https://github.com/fastavro/fastavro/issues/415
+            schema_name = parsed_schema.get("name", schema_dict["type"])
 
         self._schema = schema
         self._schema_name = schema_name

tests/integration/schema_registry/data/union_schema.avsc

@@ -0,0 +1,22 @@
+[
+    {
+        "name": "RecordOne",
+        "type": "record",
+        "fields": [
+            {
+                "name": "field_one",
+                "type": "string"
+            }
+        ]
+    },
+    {
+        "name": "RecordTwo",
+        "type": "record",
+        "fields": [
+            {
+                "name": "field_two",
+                "type": "int"
+            }
+        ]
+    }
+]

tests/schema_registry/test_avro_serializer.py

@@ -34,7 +34,7 @@ def test_avro_serializer_config_auto_register_schemas():
     """
     conf = {'url': TEST_URL}
     test_client = SchemaRegistryClient(conf)
-    test_serializer = AvroSerializer(test_client, 'string',
+    test_serializer = AvroSerializer(test_client, '"string"',
                                      conf={'auto.register.schemas': False})
     assert not test_serializer._auto_register
 
@@ -60,7 +60,7 @@ def test_avro_serializer_config_auto_register_schemas_false(mock_schema_registry
     topic = "test-auto-register"
     subject = topic + '-key'
 
-    test_serializer = AvroSerializer(test_client, 'string',
+    test_serializer = AvroSerializer(test_client, '"string"',
                                      conf={'auto.register.schemas': False})
 
     test_serializer("test",
@@ -83,7 +83,7 @@ def test_avro_serializer_config_use_latest_version(mock_schema_registry):
     topic = "test-use-latest-version"
     subject = topic + '-key'
 
-    test_serializer = AvroSerializer(test_client, 'string',
+    test_serializer = AvroSerializer(test_client, '"string"',
                                      conf={'auto.register.schemas': False, 'use.latest.version': True})
 
     test_serializer("test",
@@ -104,7 +104,7 @@ def test_avro_serializer_config_subject_name_strategy():
 
     conf = {'url': TEST_URL}
     test_client = SchemaRegistryClient(conf)
-    test_serializer = AvroSerializer(test_client, 'int',
+    test_serializer = AvroSerializer(test_client, '"int"',
                                      conf={'subject.name.strategy':
                                            record_subject_name_strategy})
 
@@ -119,7 +119,7 @@ def test_avro_serializer_config_subject_name_strategy_invalid():
     conf = {'url': TEST_URL}
     test_client = SchemaRegistryClient(conf)
     with pytest.raises(ValueError, match="must be callable"):
-        AvroSerializer(test_client, 'int',
+        AvroSerializer(test_client, '"int"',
                        conf={'subject.name.strategy': dict()})
 
 
@@ -148,7 +148,7 @@ def test_avro_serializer_record_subject_name_strategy_primitive(load_avsc):
     """
     conf = {'url': TEST_URL}
     test_client = SchemaRegistryClient(conf)
-    test_serializer = AvroSerializer(test_client, 'int',
+    test_serializer = AvroSerializer(test_client, '"int"',
                                      conf={'subject.name.strategy':
                                            record_subject_name_strategy})
 
@@ -181,7 +181,7 @@ def test_avro_serializer_topic_record_subject_name_strategy_primitive(load_avsc)
     """
     conf = {'url': TEST_URL}
     test_client = SchemaRegistryClient(conf)
-    test_serializer = AvroSerializer(test_client, 'int',
+    test_serializer = AvroSerializer(test_client, '"int"',
                                      conf={'subject.name.strategy':
                                            topic_record_subject_name_strategy})
 
@@ -204,3 +204,20 @@ def test_avro_serializer_subject_name_strategy_default(load_avsc):
     ctx = SerializationContext('test_subj', MessageField.VALUE)
     assert test_serializer._subject_name_func(
         ctx, test_serializer._schema_name) == 'test_subj-value'
+
+
+def test_avro_serializer_schema_loads_union(load_avsc):
+    """
+    Ensures union types are correctly parsed
+    """
+    conf = {'url': TEST_URL}
+    test_client = SchemaRegistryClient(conf)
+    test_serializer = AvroSerializer(test_client,
+                                     load_avsc('union_schema.avsc'))
+
+    assert test_serializer._schema_name is None
+
+    schema = test_serializer._parsed_schema
+    assert isinstance(schema, list)
+    assert schema[0]["name"] == "RecordOne"
+    assert schema[1]["name"] == "RecordTwo"