@Override
    public GenericAvroRecord read(byte[] bytes, int offset, int length) {
        try {
            if (offset == 0 && this.offset > 0) {
                offset = this.offset;
            }
            Decoder decoder = DecoderFactory.get().binaryDecoder(bytes, offset, length - offset, null);
            org.apache.avro.generic.GenericRecord avroRecord =
                    (org.apache.avro.generic.GenericRecord) reader.read(
                    null,
                    decoder);
            return new GenericAvroRecord(schemaVersion, schema, fields, avroRecord);
        } catch (IOException | IndexOutOfBoundsException e) {
            throw new SchemaSerializationException(e);
        }
    }
@Test
    public void testGenericAvroReaderByReaderSchema() {
        byte[] fooV2Bytes = fooV2Schema.encode(fooV2);

        GenericAvroReader genericAvroSchemaByReaderSchema = new GenericAvroReader(fooV2Schema.getAvroSchema(), fooSchemaNotNull.getAvroSchema(), new byte[10]);
        GenericRecord genericRecordByReaderSchema = genericAvroSchemaByReaderSchema.read(fooV2Bytes);
        assertEquals(genericRecordByReaderSchema.getField("fieldUnableNull"), "defaultValue");
        assertEquals(genericRecordByReaderSchema.getField("field1"), "foo1");
        assertEquals(genericRecordByReaderSchema.getField("field3"), 10);
    }