Union type can't be serialized

[keremsahin1]

Description

I'm trying to work out the exact same example for AvroProducer with a slightly different schema and value as below.

from confluent_kafka import avro
from confluent_kafka.avro import AvroProducer


value_schema_str = """
{
   "namespace": "my.test",
   "name": "example_value_schema",
   "type": "record",
   "fields" : [
     {
       "name" : "field",
       "type" : [ "null", 
         {
           "type": "record",
           "name": "my_field_type",
           "fields": [ {
             "name": "subfield",
             "type": "string"
           } ]
         }
       ]
     }
   ]
}
"""

key_schema_str = """
{
   "namespace": "my.test",
   "name": "key",
   "type": "record",
   "fields" : [
     {
       "name" : "name",
       "type" : "string"
     }
   ]
}
"""

value_schema = avro.loads(value_schema_str)
key_schema = avro.loads(key_schema_str)
value = {"field": {"my_field_type": {"subfield": "test"}}}
key = {"name": "Key"}

avroProducer = AvroProducer({
    'bootstrap.servers': 'localhost:9092',
    'schema.registry.url': 'http://localhost:8081'
    }, default_key_schema=key_schema, default_value_schema=value_schema)

avroProducer.produce(topic='my_topic2', value=value, key=key)
avroProducer.flush()

And I get below error:

Traceback (most recent call last):
File "producer2.py", line 52, in
avroProducer.produce(topic='my_topic2', value=value, key=key)
File "/Library/Python/2.7/site-packages/confluent_kafka/avro/init.py", line 80, in produce
value = self._serializer.encode_record_with_schema(topic, value_schema, value)
File "/Library/Python/2.7/site-packages/confluent_kafka/avro/serializer/message_serializer.py", line 113, in encode_record_with_schema
return self.encode_record_with_schema_id(schema_id, record, is_key=is_key)
File "/Library/Python/2.7/site-packages/confluent_kafka/avro/serializer/message_serializer.py", line 147, in encode_record_with_schema_id
writer(record, outf)
File "/Library/Python/2.7/site-packages/confluent_kafka/avro/serializer/message_serializer.py", line 82, in
return lambda record, fp: schemaless_writer(fp, schema, record)
File "fastavro/_write.pyx", line 582, in fastavro._write.schemaless_writer
File "fastavro/_write.pyx", line 335, in fastavro._write.write_data
File "fastavro/_write.pyx", line 285, in fastavro._write.write_record
File "fastavro/_write.pyx", line 333, in fastavro._write.write_data
File "fastavro/_write.pyx", line 249, in fastavro._write.write_union
ValueError: {'my_field_type': {'subfield': 'test'}} (type <type 'dict'>) do not match [u'null', {'fields': [{'type': u'string', 'name': u'subfield'}], 'type': u'record', 'name': 'my.test.my_field_type'}]

However, if I change my value to value = {"field": {"subfield": "test"}} it seems to work. Why is this not supported? What happens if I define two union records with exact same schema?

Checklist

Please provide the following information:

• confluent-kafka-python and librdkafka version (confluent_kafka.version() and confluent_kafka.libversion()):
• Apache Kafka broker version:
• Client configuration: {...}
• Operating system:
• Provide client logs (with 'debug': '..' as necessary)
• Provide broker log excerpts
• Critical issue

[scottbelden]

The schema you have defined is a record with a key called field and a value of either null or a my_field_type record. So the record should look like one of these two objects:

{"field": None}
# or
{"field": <my_field_type>}

The my_field_type record is defined as an object with a key of subfield and a value of a string type. So something like this: {"subfield": "some_string"}. If we substitute that into the placeholder from above, the record for your schema should have the following structure:

{"field": None}
# or
{"field": {"subfield": "some_string"}}

So, your second example of value = {"field": {"subfield": "test"}} works because it conforms to the schema and the first example of value = {"field": {"my_field_type": {"subfield": "test"}}} doesn't work because it doesn't conform to the schema. You can call the record whatever name you want, but either way, that name isn't used in the actual data structure.

Does this make sense?

[keremsahin1]

@scottbelden Thanks for the detailed explanation.
What happens if I define a schema as below:

value_schema_str = """
{
  "namespace": "my.test",
  "name": "example_value_schema",
  "type": "record",
  "fields": [{
    "name": "field",
    "type": ["null",
      {
        "type": "record",
        "name": "my_field_type1",
        "fields": [{
          "name": "subfield",
          "type": "string"
        }]
      },
      {
        "type": "record",
        "name": "my_field_type2",
        "fields": [{
          "name": "subfield",
          "type": "string"
        }]
      }
    ]
  }]
}
"""

In this example, both my_field_type1 and my_field_type2 has the same schema. How does it get resolved which type am I using in this case? Isn't there a way to hint types to Avro?

[scottbelden]

I believe the way this project works is that it will try to use fastavro and if for some reason that fails it will fall back to the official avro library.

In fastavro it possible to do what you are trying to do. To do it, the value of the union would be a tuple of (type, data) instead of just data. For example, in your schema above, to specify that you want the data to be a my_field_type1, you would have something like this:

{"field": ("my_field_type1", {"subfield": "some_string"})}

However, this functionality doesn't exist in the fallback avro library. So if for some reason fastavro failed during the serialization/deserialization, then avro would definitely fail. However, this is already the case for some other aspects because there are a few features that fastavro supports that the official avro does not.

Finally, note that this tuple syntax is only used for serializing and is not generated when deserializing. So if you were to read the avro binary, you would get the following returned to you:

{"field": {"subfield": "some_string"}}

There's an outstanding issue that talks about this issue here: fastavro/fastavro#191

[slominskir]

Isn't there a way to hint types to Avro?

As Scott mentioned using a tuple works so this ticket can probably be closed. Wouldn't hurt to document this behavior in the confluent-kafka-python docs somewhere though.

[slominskir]

@scottbelden -

It probably is worth mentioning that the tuple notation does make compatibility with the AVRO spec and the Confluent kafka-avro-console-consumer tricky. The on-the-wire AVRO protocol format is fine, but the AVRO standard also specifies how to encode AVRO in JSON for human consumption and it isn't a tuple - it's all JSON. Is this something fastavro could/should handle?

[scottbelden]

The tuple notation is not an output and doesn't show up in the JSON encoded format. To see that, we can run this script using one of the schemas from above that are ambiguous:

from fastavro import json_writer
from io import StringIO

schema = {
    "namespace": "my.test",
    "name": "example_value_schema",
    "type": "record",
    "fields": [
        {
            "name": "field",
            "type": [
                "null",
                {
                    "type": "record",
                    "name": "my_field_type1",
                    "fields": [{"name": "subfield", "type": "string"}]
                },
                {
                    "type": "record",
                    "name": "my_field_type2",
                    "fields": [{"name": "subfield", "type": "string"}]
                }
            ]
        }
   ]
}

sio = StringIO()
records = [
    # `null` union option
    {"field": None},

    # Do not specify a union path. Fastavro will try to figuire out union branch
    # is correct. In this case, `my_field_type1` and `my_field_type2` are the
    # same so it will not know for sure which path is the correct one and will
    # go with the first option: my_field_type1
    {"field": {"subfield": "foo"}},

    # Specify the union path as `my_field_type2`. This tells fastavro that the
    # ambiguous subrecord is actually a `my_field_type2`.
    {"field": ("my.test.my_field_type2", {"subfield": "foo"})},
]

json_writer(sio, schema, records)

print(sio.getvalue())

The output of that script is:

{"field": null}
{"field": {"my.test.my_field_type1": {"subfield": "foo"}}}
{"field": {"my.test.my_field_type2": {"subfield": "foo"}}}

The JSON encoded output format doesn't include any tuples.

[slominskir]

So it sounds like the json_writer and json_reader know how to handle union tuples properly, but they don't appear to handle schema references as there is no named_schemas parameter so if I pass a schema string containing references it blows up with:

fastavro._schema_common.UnknownType

[scottbelden]

Do you have an example of the failure? If I go back to the script I posted but alter it some to have a sub schema that is broken out from the main schema then it does seem to work for me:

from fastavro import json_writer, parse_schema
from io import StringIO

sub_schema = {
    "namespace": "my.test",
    "type": "record",
    "name": "my_field_type1",
    "fields": [{"name": "subfield", "type": "string"}]
}

named_schemas = {}
parsed_sub_schema = parse_schema(sub_schema, named_schemas)

schema = {
    "namespace": "my.test",
    "name": "example_value_schema",
    "type": "record",
    "fields": [
        {
            "name": "field",
            "type": [
                "null",
                "my.test.my_field_type1",
                {
                    "type": "record",
                    "name": "my_field_type2",
                    "fields": [{"name": "subfield", "type": "string"}]
                }
            ]
        }
   ]
}

parsed_schema = parse_schema(schema, named_schemas)

sio = StringIO()
records = [
    # `null` union option
    {"field": None},

    # Do not specify a union path. Fastavro will try to figuire out union branch
    # is correct. In this case, `my_field_type1` and `my_field_type2` are the
    # same so it will not know for sure which path is the correct one and will
    # go with the first option: my_field_type1
    {"field": {"subfield": "foo"}},

    # Specify the union path as `my_field_type2`. This tells fastavro that the
    # ambiguous subrecord is actually a `my_field_type2`.
    {"field": ("my.test.my_field_type2", {"subfield": "foo"})},
]

json_writer(sio, parsed_schema, records)

print(sio.getvalue())

[slominskir]

Is overloading the second positional parameter of json_writer to accept either a String or Schema object (dict) good form in Python? Seems surprising and confusing. Should kwargs be used instead? I guess the support for human readable JSON AVRO is there though - my hang up was that I leveraging the fact that print dictionary results in JSON, but since unions are represented as tuples you almost get good JSON just printing the dictionary except for unions.

[scottbelden]

The second parameter of json_writer is always a python dictionary. In the case of fastavro, unlike the avro library, a "parsed" schema is also a python dictionary, not anything fancy like a python class.

And yes, you are right that the Avro JSON output is pretty close to the internal dictionary representation of a record in python with unions being the big difference. However, the other big difference that will catch people is any fields that are a bytes type (or also any logical types).

For example, while you could get an Avro JSON for a record like this:

import json
schema = {
    "namespace": "my.test",
    "name": "example_value_schema",
    "type": "record",
    "fields": [
        {"name": "field1", "type": "null"},
        {"name": "field2", "type": "boolean"},
        {"name": "field3", "type": "string"},
   ]
}

record = {
    "field1": None,
    "field2": True,
    "field3": "foo",
}

print(json.dumps(record))

It would fail if you had a field with a bytes type:

import json

schema = {
    "namespace": "my.test",
    "name": "example_value_schema",
    "type": "record",
    "fields": [
        {"name": "field1", "type": "null"},
        {"name": "field2", "type": "boolean"},
        {"name": "field3", "type": "string"},
        {"name": "field4", "type": "bytes"},
   ]
}

record = {
    "field1": None,
    "field2": True,
    "field3": "foo",
    "field4": b"bar",
}

print(json.dumps(record))

[slominskir]

OK, looks like AVRO JSON format is supported - just use json_writer and json_reader.