### Question:
RegistryAvroDeserializationSchema extends AvroDeserializationSchema<T> { @Override public T deserialize(byte[] message) throws IOException { checkAvroInitialized(); getInputStream().setBuffer(message); Schema writerSchema = schemaCoder.readSchema(getInputStream()); Schema readerSchema = getReaderSchema(); GenericDatumReader<T> datumReader = getDatumReader(); datumReader.setSchema(writerSchema); datumReader.setExpected(readerSchema); return datumReader.read(null, getDecoder()); } protected  RegistryAvroDeserializationSchema(Class<T> recordClazz, @Nullable Schema reader,
			SchemaCoder.SchemaCoderProvider schemaCoderProvider); @Override T deserialize(byte[] message);  }### Answer:
@Test public 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(); } } ); GenericRecord genericRecord = deserializer.deserialize(writeRecord( address, Address.getClassSchema())); assertEquals(address.getNum(), genericRecord.get("num")); assertEquals(address.getStreet(), genericRecord.get("street").toString()); assertNull(genericRecord.get("city")); assertNull(genericRecord.get("state")); assertNull(genericRecord.get("zip")); assertNull(genericRecord.get("country")); }
@Test public void testSpecificRecordReadMoreFieldsThanWereWritten() throws IOException { Schema smallerUserSchema = new Schema.Parser().parse( "{\"namespace\": \"org.apache.flink.formats.avro.generated\",\n" + " \"type\": \"record\",\n" + " \"name\": \"SimpleRecord\",\n" + " \"fields\": [\n" + " {\"name\": \"name\", \"type\": \"string\"}" + " ]\n" + "}]"); RegistryAvroDeserializationSchema<SimpleRecord> deserializer = new RegistryAvroDeserializationSchema<>( SimpleRecord.class, null, () -> in -> smallerUserSchema ); GenericData.Record smallUser = new GenericRecordBuilder(smallerUserSchema) .set("name", "someName") .build(); SimpleRecord simpleRecord = deserializer.deserialize(writeRecord( smallUser, smallerUserSchema)); assertEquals("someName", simpleRecord.getName().toString()); assertNull(simpleRecord.getOptionalField()); }