@Override
    public GenericProtobufNativeRecord read(byte[] bytes, int offset, int length) {
        try {
            if (!(bytes.length == length && offset == 0)) { //skip unnecessary bytes copy
                bytes = Arrays.copyOfRange(bytes, offset, offset + length);
            }
            return new GenericProtobufNativeRecord(schemaVersion, descriptor, fields,
                    DynamicMessage.parseFrom(descriptor, bytes));
        } catch (InvalidProtocolBufferException e) {
            throw new SchemaSerializationException(e);
        }
    }
@Test
    public void testClazzBasedReaderByGenericWriterSchema() {
        genericmessage = genericProtobufNativeSchema.newRecordBuilder().set("stringField", STRING_FIELD_VLUE).set("doubleField", DOUBLE_FIELD_VLUE).build();
        byte[] messageBytes = new GenericProtobufNativeWriter().write(genericmessage);
        GenericProtobufNativeReader genericProtobufNativeReader = new GenericProtobufNativeReader(clazzBasedProtobufNativeSchema.getProtobufNativeSchema());
        GenericRecord genericRecordByWriterSchema = genericProtobufNativeReader.read(messageBytes);
        assertEquals(genericRecordByWriterSchema.getField("stringField"), STRING_FIELD_VLUE);
        assertEquals(genericRecordByWriterSchema.getField("doubleField"), DOUBLE_FIELD_VLUE);

    }