@Override
    public Mutation createMutation(Record<GenericObject> message) {
        final GenericObject record = message.getValue();
        Function<String, Object> recordValueGetter;
        MutationType mutationType = null;
        if (message.getSchema() != null && message.getSchema() instanceof KeyValueSchema) {
            KeyValueSchema<GenericObject, GenericObject> keyValueSchema = (KeyValueSchema) message.getSchema();

            final org.apache.pulsar.client.api.Schema<GenericObject> keySchema = keyValueSchema.getKeySchema();
            final org.apache.pulsar.client.api.Schema<GenericObject> valueSchema = keyValueSchema.getValueSchema();
            KeyValue<GenericObject, GenericObject> keyValue =
                    (KeyValue<GenericObject, GenericObject>) record.getNativeObject();

            final GenericObject key = keyValue.getKey();
            final GenericObject value = keyValue.getValue();

            boolean isDelete = false;
            if (value == null) {
                switch (jdbcSinkConfig.getNullValueAction()) {
                    case DELETE:
                        isDelete = true;
                        break;
                    case FAIL:
                        throw new IllegalArgumentException("Got record with value NULL with nullValueAction=FAIL");
                    default:
                        break;
                }
            }
            Map<String, Object> data = new HashMap<>();
            fillKeyValueSchemaData(keySchema, key, data);
            if (isDelete) {
                mutationType = MutationType.DELETE;
            } else {
                fillKeyValueSchemaData(valueSchema, value, data);
            }
            recordValueGetter = (k) -> data.get(k);
        } else {
            SchemaType schemaType = message.getSchema().getSchemaInfo().getType();
            if (schemaType.isPrimitive()) {
                throw new UnsupportedOperationException("Primitive schema is not supported: " + schemaType);
            }
            recordValueGetter = (key) -> ((GenericRecord) record).getField(key);
        }
        String action = message.getProperties().get(ACTION_PROPERTY);
        if (action != null) {
            mutationType = MutationType.valueOf(action);
        } else if (mutationType == null) {
            switch (jdbcSinkConfig.getInsertMode()) {
                case INSERT:
                    mutationType = MutationType.INSERT;
                    break;
                case UPSERT:
                    mutationType = MutationType.UPSERT;
                    break;
                case UPDATE:
                    mutationType = MutationType.UPDATE;
                    break;
                default:
                    throw new IllegalArgumentException("Unknown insert mode: " + jdbcSinkConfig.getInsertMode());
            }
        }
        return new Mutation(mutationType, recordValueGetter);
    }
@Test(expectedExceptions = UnsupportedOperationException.class,
            expectedExceptionsMessageRegExp = "Primitive schema is not supported.*")
    @SuppressWarnings("unchecked")
    public void testNotSupportPrimitiveSchema() {
        BaseJdbcAutoSchemaSink baseJdbcAutoSchemaSink = new BaseJdbcAutoSchemaSink() {};
        AutoConsumeSchema autoConsumeSchema = new AutoConsumeSchema();
        autoConsumeSchema.setSchema(org.apache.pulsar.client.api.Schema.STRING);
        Record<? extends GenericObject> record = new Record<GenericRecord>() {
            @Override
            public org.apache.pulsar.client.api.Schema<GenericRecord> getSchema() {
                return autoConsumeSchema;
            }

            @Override
            public GenericRecord getValue() {
                return null;
            }
        };
        baseJdbcAutoSchemaSink.createMutation((Record<GenericObject>) record);
    }