@Override
    public T deserialize(@Nullable byte[] message) throws IOException {
        if (message == null) {
            return null;
        }
        checkAvroInitialized();
        getInputStream().setBuffer(message);
        Schema writerSchema = schemaCoder.readSchema(getInputStream());
        Schema readerSchema = getReaderSchema();

        GenericDatumReader<T> datumReader = getDatumReader();

        datumReader.setSchema(writerSchema);
        datumReader.setExpected(readerSchema);

        if (getEncoding() == AvroEncoding.JSON) {
            ((JsonDecoder) getDecoder()).configure(getInputStream());
        }

        return datumReader.read(null, getDecoder());
    }
@Test
    void testGenericRecordReadWithCompatibleSchema() throws IOException {
        RegistryAvroDeserializationSchema<GenericRecord> deserializer =
                new RegistryAvroDeserializationSchema<>(
                        GenericRecord.class,
                        SchemaBuilder.record("Address")
                                .fields()
                                .requiredString("street")
                                .requiredInt("num")
                                .optionalString("country")
                                .endRecord(),
                        () ->
                                new SchemaCoder() {
                                    @Override
                                    public Schema readSchema(InputStream in) {
                                        return Address.getClassSchema();
                                    }

                                    @Override
                                    public void writeSchema(Schema schema, OutputStream out)
                                            throws IOException {
                                        // do nothing
                                    }
                                });

        GenericRecord genericRecord =
                deserializer.deserialize(writeRecord(address, Address.getClassSchema()));
        assertThat(genericRecord.get("num")).isEqualTo(address.getNum());
        assertThat(genericRecord.get("street").toString()).isEqualTo(address.getStreet());
        assertThat(genericRecord.get("country")).isNull();
    }