public static KafkaRoutineLoadJob fromCreateStmt(CreateRoutineLoadStmt stmt) throws UserException {
        // check db and table
        Database db = GlobalStateMgr.getCurrentState().getDb(stmt.getDBName());
        if (db == null) {
            ErrorReport.reportDdlException(ErrorCode.ERR_BAD_DB_ERROR, stmt.getDBName());
        }

        Table table = db.getTable(stmt.getTableName());
        if (table == null) {
            ErrorReport.reportDdlException(ErrorCode.ERR_BAD_TABLE_ERROR, stmt.getTableName());
        }

        long tableId = table.getId();
        Locker locker = new Locker();
        locker.lockTablesWithIntensiveDbLock(db, Lists.newArrayList(tableId), LockType.READ);
        try {
            unprotectedCheckMeta(db, stmt.getTableName(), stmt.getRoutineLoadDesc());
            Load.checkMergeCondition(stmt.getMergeConditionStr(), (OlapTable) table, table.getFullSchema(), false);
        } finally {
            locker.unLockTablesWithIntensiveDbLock(db, Lists.newArrayList(tableId), LockType.READ);
        }

        // init kafka routine load job
        long id = GlobalStateMgr.getCurrentState().getNextId();
        KafkaRoutineLoadJob kafkaRoutineLoadJob = new KafkaRoutineLoadJob(id, stmt.getName(),
                db.getId(), tableId, stmt.getKafkaBrokerList(), stmt.getKafkaTopic());
        kafkaRoutineLoadJob.setOptional(stmt);
        kafkaRoutineLoadJob.checkCustomProperties();

        return kafkaRoutineLoadJob;
    }
@Test
    public void testSerializationJson(@Mocked GlobalStateMgr globalStateMgr,
                                      @Injectable Database database,
                                      @Injectable OlapTable table) throws UserException {
        CreateRoutineLoadStmt createRoutineLoadStmt = initCreateRoutineLoadStmt();
        Map<String, String> jobProperties = createRoutineLoadStmt.getJobProperties();
        jobProperties.put("format", "json");
        jobProperties.put("strip_outer_array", "true");
        jobProperties.put("jsonpaths", "['$.category','$.price','$.author']");
        jobProperties.put("json_root", "");
        jobProperties.put("timezone", "Asia/Shanghai");
        createRoutineLoadStmt.checkJobProperties();

        RoutineLoadDesc routineLoadDesc = new RoutineLoadDesc(columnSeparator, null, null, null, partitionNames);
        Deencapsulation.setField(createRoutineLoadStmt, "routineLoadDesc", routineLoadDesc);
        List<Pair<Integer, Long>> partitionIdToOffset = Lists.newArrayList();
        for (String s : kafkaPartitionString.split(",")) {
            partitionIdToOffset.add(new Pair<>(Integer.valueOf(s), 0L));
        }
        Deencapsulation.setField(createRoutineLoadStmt, "kafkaPartitionOffsets", partitionIdToOffset);
        Deencapsulation.setField(createRoutineLoadStmt, "kafkaBrokerList", serverAddress);
        Deencapsulation.setField(createRoutineLoadStmt, "kafkaTopic", topicName);
        long dbId = 1L;
        long tableId = 2L;

        new Expectations() {
            {
                database.getTable(tableNameString);
                minTimes = 0;
                result = table;
                database.getId();
                minTimes = 0;
                result = dbId;
                table.getId();
                minTimes = 0;
                result = tableId;
                table.isOlapOrCloudNativeTable();
                minTimes = 0;
                result = true;
            }
        };

        new MockUp<KafkaUtil>() {
            @Mock
            public List<Integer> getAllKafkaPartitions(String brokerList, String topic,
                                                       ImmutableMap<String, String> properties) throws UserException {
                return Lists.newArrayList(1, 2, 3);
            }
        };

        String createSQL = "CREATE ROUTINE LOAD db1.job1 ON table1 " +
                "PROPERTIES('format' = 'json', 'strip_outer_array' = 'true') " +
                "FROM KAFKA('kafka_broker_list' = 'http://127.0.0.1:8080','kafka_topic' = 'topic1');";
        KafkaRoutineLoadJob job = KafkaRoutineLoadJob.fromCreateStmt(createRoutineLoadStmt);
        job.setOrigStmt(new OriginStatement(createSQL, 0));
        Assert.assertEquals("json", job.getFormat());
        Assert.assertTrue(job.isStripOuterArray());
        Assert.assertEquals("['$.category','$.price','$.author']", job.getJsonPaths());
        Assert.assertEquals("", job.getJsonRoot());

        String data = GsonUtils.GSON.toJson(job, KafkaRoutineLoadJob.class);
        KafkaRoutineLoadJob newJob = GsonUtils.GSON.fromJson(data, KafkaRoutineLoadJob.class);
        Assert.assertEquals("json", newJob.getFormat());
        Assert.assertTrue(newJob.isStripOuterArray());
        Assert.assertEquals("['$.category','$.price','$.author']", newJob.getJsonPaths());
        Assert.assertEquals("", newJob.getJsonRoot());
    }