ClassCastException on Nullable Map Value Schemas with Backward Compatible Changes (Field Addition)

java.lang.ClassCastException: org.apache.avro.Resolver$ReaderUnion cannot be cast to org.apache.avro.Resolver$Container

	at org.apache.avro.io.parsing.ResolvingGrammarGenerator.generate(ResolvingGrammarGenerator.java:99)
	at org.apache.avro.io.parsing.ResolvingGrammarGenerator.generate(ResolvingGrammarGenerator.java:110)
	at org.apache.avro.io.parsing.ResolvingGrammarGenerator.generate(ResolvingGrammarGenerator.java:141)
	at org.apache.avro.io.parsing.ResolvingGrammarGenerator.generate(ResolvingGrammarGenerator.java:65)
	at org.apache.avro.io.ResolvingDecoder.resolve(ResolvingDecoder.java:85)
	at org.apache.avro.io.ResolvingDecoder.<init>(ResolvingDecoder.java:50)
	at org.apache.avro.io.DecoderFactory.resolvingDecoder(DecoderFactory.java:297)
	at org.apache.avro.generic.GenericDatumReader.getResolver(GenericDatumReader.java:128)
	at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:142)

import java.io.ByteArrayOutputStream;
import java.util.Collections;

import org.apache.avro.Schema;
import org.apache.avro.io.DecoderFactory;
import org.apache.avro.io.EncoderFactory;
import org.apache.avro.reflect.ReflectData;
import org.apache.avro.reflect.ReflectDatumWriter;
import org.junit.Test;

import static org.junit.Assert.assertEquals;

public class AvroUpgradeTest {

    @Test
    public void nullableMapWithBackwardCompatibleValueChangesCanBeDeserialized() throws Exception {
    TestDataWithNullableMap data = new TestDataWithNullableMap();
    data.setMap(Collections.singletonMap("KEY", TestData.create()));

    Schema writerSchema = ReflectData.get().getSchema(TestDataWithNullableMap.class);

    Schema readerMapSchema = Schema.createMap(Schema.createRecord(TestData.class.getName(), null, null, false,
        Collections.singletonList(new Schema.Field("data1", Schema.create(Schema.Type.STRING), null, Object.class.cast(null)))));
    Schema readerSchema = Schema.createRecord(TestDataWithNullableMap.class.getName(), null, null, false,
        Collections.singletonList(new Schema.Field("map", ReflectData.makeNullable(readerMapSchema), null, Object.class.cast(null))));

    ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
    new ReflectDatumWriter<>(writerSchema, ReflectData.get()).write(data, EncoderFactory.get().directBinaryEncoder(outputStream, null));

    byte[] serialized = outputStream.toByteArray();

    TestDataWithNullableMap deserialized = new ReflectDecoderAvroDeserializer.ReflectDecoderDatumReader<TestDataWithNullableMap>(writerSchema, readerSchema, ReflectData.get())
        .read(null, DecoderFactory.get().binaryDecoder(serialized, 0, serialized.length, null));

    assertEquals(data.getMap().get("KEY").getData1(), deserialized.getMap().get("KEY").getData1());
    }
}

import java.util.Map;

import org.apache.avro.reflect.Nullable;

public class TestDataWithNullableMap {

    @Nullable
    private Map<String, TestData> map;

    public Map<String, TestData> getMap() {
        return map;
    }

    public void setMap(Map<String, TestData> map) {
        this.map = map;
    }
}

public class TestData {

    private String data1;

    private String data2;

    public static TestData create() {
        TestData data = new TestData();
        data.setData1("DATA 1");
        data.setData2("DATA 2");
        return data;
    }

    public String getData1() {
        return data1;
    }

    public void setData1(String data1) {
        this.data1 = data1;
    }

    public String getData2() {
        return data2;
    }

    public void setData2(String data2) {
        this.data2 = data2;
    }
}