@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 testGetNativeRecord() {
        message = TestMessage.newBuilder().setStringField(STRING_FIELD_VLUE).setDoubleField(DOUBLE_FIELD_VLUE).build();
        GenericProtobufNativeReader genericProtobufNativeReader = new GenericProtobufNativeReader(genericProtobufNativeSchema.getProtobufNativeSchema());
        GenericRecord record = genericProtobufNativeReader.read(message.toByteArray());
        assertEquals(record.getField("stringField"), STRING_FIELD_VLUE);
        assertEquals(record.getField("doubleField"), DOUBLE_FIELD_VLUE);
        assertEquals(SchemaType.PROTOBUF_NATIVE, record.getSchemaType());
        DynamicMessage nativeRecord = (DynamicMessage) record.getNativeObject();
        assertEquals(nativeRecord.getField(nativeRecord.getDescriptorForType().findFieldByName("stringField")), STRING_FIELD_VLUE);
        assertEquals(nativeRecord.getField(nativeRecord.getDescriptorForType().findFieldByName("doubleField")), DOUBLE_FIELD_VLUE);
    }