@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()); }
@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()); }
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()); } }
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); }
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); }
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);  }