Pulsar: Using Map's with Avro Generated Types Generates Weird Errors

[riferrei]

Description of the bug
I have a project that uses Avro generated types (POJO's generated by the Avro compiler) as the type to be serialized with the producer. Everything works fine excepts if one of the fields is a map. When the field is a map and a value is set into this map, the following error is generated:

org.apache.pulsar.shade.org.apache.avro.UnresolvedUnionException: Not in union
   ["null",{"type":"array","items":{"type":"record","name":"PairCharSequenceDouble",
   "namespace":"org.apache.pulsar.shade.org.apache.avro.reflect",
   "fields":[{"name":"key","type":"string"},{"name":"value","type":"double"}]},
   "java-class":"java.util.Map"}]

Anecdotally, if the field containing a map is not set, the message is properly serialized, proving that the problem is indeed the map field. Also, the same type if implemented as a POJO class by hand (without using a Avro compiler) works fine. Which means that something happens during the message serialization when the type being serialized contains the code generated by the compiler.

To Reproduce
Steps to reproduce the behavior:

1. Define an new type like the one below:

{
    "namespace": "com.acme.something",
    "type": "record",
    "name": "AvroGenComplexType",
    "fields": [
        {
            "name": "stringField",
            "type": "string"
        },
        {
            "name": "booleanField",
            "type": "boolean"
        },
        {
            "name": "bytesField",
            "type": "bytes"
        },
        {
            "name": "intField",
            "type": "int"
        },
        {
            "name": "longField",
            "type": "long"
        },
        {
            "name": "floatField",
            "type": "float"
        },
        {
            "name": "doubleField",
            "type": "double"
        },
        {
            "name": "mapField",
            "type": [
                "null",
                {
                    "type": "map",
                    "values": "double"
                }
            ],
            "default": null
        },
        {
            "name": "innerField",
            "type": [
                "null",
                {
                    "type": "record",
                    "name": "AvroGenInnerType",
                    "fields": [
                        {
                            "name": "doubleField",
                            "type": "double"
                        },
                        {
                            "name": "arrayField",
                            "type": {
                                "type": "array",
                                "items": "string"
                            }
                        },
                        {
                            "name": "enumField",
                            "type": {
                                "type": "enum",
                                "name": "AvroGenMultipleOptions",
                                "symbols": [
                                    "FirstOption", "SecondOption",
                                    "ThirdOption", "FourthOption"
                                ],
                                "default": "FirstOption"
                            }
                        }
                    ]
                }
            ],
            "default": null
        }
    ]
}

1. Using a Avro compiler, generate the types for Java.
2. Write a Pulsar producer with the following code:

    protected AvroGenComplexType createAvroGenComplexType() {

        AvroGenComplexType.Builder complexType =
            AvroGenComplexType.newBuilder();

        complexType.setStringField(String.valueOf(RANDOM.nextInt()));
        complexType.setBooleanField(RANDOM.nextBoolean());
        byte[] bytes = new byte[1024];
        RANDOM.nextBytes(bytes);
        ByteBuffer byteBuffer = ByteBuffer.wrap(bytes);
        complexType.setBytesField(byteBuffer);
        complexType.setIntField(RANDOM.nextInt());
        complexType.setLongField(RANDOM.nextLong());
        complexType.setFloatField(RANDOM.nextFloat());
        complexType.setDoubleField(RANDOM.nextDouble());

        // ******** This part will generate the error ********
        Map<CharSequence, Double> mapField = new HashMap<>(1);
        mapField.put(String.valueOf(RANDOM.nextInt()), RANDOM.nextDouble());
        complexType.setMapField(mapField);
        // *****************************************************

        complexType.setInnerFieldBuilder(AvroGenInnerType.newBuilder()
            .setDoubleField(RANDOM.nextDouble())
            .setArrayField(Arrays.asList(String.valueOf(RANDOM.nextInt())))
            .setEnumField(AvroGenMultipleOptions.ThirdOption));

        return complexType.build();

    }

    public void produceAvroGenBasedMessages(String topic, int numMessages)
        throws PulsarClientException {
        Producer<AvroGenComplexType> producer = null;
        try {
            producer = pulsarClient.newProducer(
                Schema.AVRO(AvroGenComplexType.class))
                    .topic(topic)
                    .create();
            for (int i = 0; i < numMessages; i++) {
                producer.send(createAvroGenComplexType());
            }
        } finally {
            if (producer != null) {
                producer.closeAsync();
            }
        }
    }

1. Execute the code and observe the error being thrown.

Expected behavior
The code should execute completely without complaining about any errors related to the type being serialized. This is true specifically because the type being used was generated by the Avro compiler, which double-checks any compliance with the spec.

Screenshots
N/A

Desktop (please complete the following information):

• OS: Tested on Linux (Fedora 31)

[codelipenghui]

Which version of the Avro compiler you are using? Pulsar uses Avro 1.8.2 and upgrades to 1.9.1 in version 2.5.1(https://github.com/apache/pulsar/pull/5938). I'm not sure is it caused by inconsistent Avro versions.

[riferrei]

I was using 1.9.2 by the time I filed the issue. However, I just tested here with both 1.8.2 and 1.9.1 and I still get the same error:

Caused by: org.apache.pulsar.shade.org.apache.avro.UnresolvedUnionException: Not in union ["null",{"type":"array","items":{"type":"record","name":"PairCharSequenceDouble","namespace":"org.apache.pulsar.shade.org.apache.avro.reflect","fields":[{"name":"key","type":"string"},{"name":"value","type":"double"}]},"java-class":"java.util.Map"}]: {-55846986=0.5533789443336609}
        at org.apache.pulsar.shade.org.apache.avro.generic.GenericData.resolveUnion(GenericData.java:740)
        at org.apache.pulsar.shade.org.apache.avro.generic.GenericDatumWriter.resolveUnion(GenericDatumWriter.java:205)
        at org.apache.pulsar.shade.org.apache.avro.generic.GenericDatumWriter.writeWithoutConversion(GenericDatumWriter.java:123)
        at org.apache.pulsar.shade.org.apache.avro.generic.GenericDatumWriter.write(GenericDatumWriter.java:75)
        at org.apache.pulsar.shade.org.apache.avro.reflect.ReflectDatumWriter.write(ReflectDatumWriter.java:159)
        at org.apache.pulsar.shade.org.apache.avro.generic.GenericDatumWriter.writeField(GenericDatumWriter.java:166)
        at org.apache.pulsar.shade.org.apache.avro.specific.SpecificDatumWriter.writeField(SpecificDatumWriter.java:90)
        at org.apache.pulsar.shade.org.apache.avro.reflect.ReflectDatumWriter.writeField(ReflectDatumWriter.java:191)
        at org.apache.pulsar.shade.org.apache.avro.generic.GenericDatumWriter.writeRecord(GenericDatumWriter.java:156)
        at org.apache.pulsar.shade.org.apache.avro.generic.GenericDatumWriter.writeWithoutConversion(GenericDatumWriter.java:118)
        at org.apache.pulsar.shade.org.apache.avro.generic.GenericDatumWriter.write(GenericDatumWriter.java:75)
        at org.apache.pulsar.shade.org.apache.avro.reflect.ReflectDatumWriter.write(ReflectDatumWriter.java:159)
        at org.apache.pulsar.shade.org.apache.avro.generic.GenericDatumWriter.write(GenericDatumWriter.java:62)
        at org.apache.pulsar.client.impl.schema.writer.AvroWriter.write(AvroWriter.java:44)

[codelipenghui]

@riferrei The problem is fixed by #6406, I have tested your reproduce demo(with 1.9.2 compiler) on the master branch and it passed. #6406 will release at 2.5.1 and 2.5.1-RC is out for validation. If you are interested in the validation you can take a look at the email thread https://lists.apache.org/thread.html/r7b2783d13007e621e2e880a37d808e6a9ec178ece2090e117b8c68e6%40%3Cdev.pulsar.apache.org%3E, thanks.

[riferrei]

Hi @codelipenghui,

Thank you so much for looking into this. I will be taking 2.5.1-RC for a spin for sure.

-- @riferrei

[codelipenghui]

@riferrei I close the issue first. If the issue not resolved, feel free to reopen it.