@Override
    protected void runAfterCatalogReady() {
        try {
            process();
        } catch (Throwable e) {
            LOG.warn("Failed to process one round of RoutineLoadScheduler", e);
        }
    }
@Test
    public void testNormalRunOneCycle(@Mocked GlobalStateMgr globalStateMgr,
                                      @Injectable RoutineLoadMgr routineLoadManager,
                                      @Injectable SystemInfoService systemInfoService,
                                      @Injectable Database database,
                                      @Injectable RoutineLoadDesc routineLoadDesc,
                                      @Mocked StreamLoadPlanner planner,
                                      @Injectable OlapTable olapTable)
            throws LoadException, MetaNotFoundException {
        List<Long> beIds = Lists.newArrayList();
        beIds.add(1L);
        beIds.add(2L);

        List<Integer> partitions = Lists.newArrayList();
        partitions.add(100);
        partitions.add(200);
        partitions.add(300);

        RoutineLoadTaskScheduler routineLoadTaskScheduler = new RoutineLoadTaskScheduler(routineLoadManager);
        Deencapsulation.setField(globalStateMgr, "routineLoadTaskScheduler", routineLoadTaskScheduler);

        KafkaRoutineLoadJob kafkaRoutineLoadJob = new KafkaRoutineLoadJob(1L, "test", 1L, 1L,
                "xxx", "test");
        Deencapsulation.setField(kafkaRoutineLoadJob, "state", RoutineLoadJob.JobState.NEED_SCHEDULE);
        List<RoutineLoadJob> routineLoadJobList = new ArrayList<>();
        routineLoadJobList.add(kafkaRoutineLoadJob);

        Deencapsulation.setField(kafkaRoutineLoadJob, "customKafkaPartitions", partitions);
        Deencapsulation.setField(kafkaRoutineLoadJob, "desireTaskConcurrentNum", 3);

        new Expectations() {
            {
                globalStateMgr.getRoutineLoadMgr();
                minTimes = 0;
                result = routineLoadManager;
                routineLoadManager.getRoutineLoadJobByState(Sets.newHashSet(RoutineLoadJob.JobState.NEED_SCHEDULE));
                minTimes = 0;
                result = routineLoadJobList;
                globalStateMgr.getDb(anyLong);
                minTimes = 0;
                result = database;
                database.getTable(1L);
                minTimes = 0;
                result = olapTable;
                systemInfoService.getBackendIds(true);
                minTimes = 0;
                result = beIds;
                routineLoadManager.getSizeOfIdToRoutineLoadTask();
                minTimes = 0;
                result = 1;
            }
        };

        RoutineLoadScheduler routineLoadScheduler = new RoutineLoadScheduler();
        Deencapsulation.setField(routineLoadScheduler, "routineLoadManager", routineLoadManager);
        routineLoadScheduler.runAfterCatalogReady();

        List<RoutineLoadTaskInfo> routineLoadTaskInfoList =
                Deencapsulation.getField(kafkaRoutineLoadJob, "routineLoadTaskInfoList");
        for (RoutineLoadTaskInfo routineLoadTaskInfo : routineLoadTaskInfoList) {
            KafkaTaskInfo kafkaTaskInfo = (KafkaTaskInfo) routineLoadTaskInfo;
            if (kafkaTaskInfo.getPartitions().size() == 2) {
                Assert.assertTrue(kafkaTaskInfo.getPartitions().contains(100));
                Assert.assertTrue(kafkaTaskInfo.getPartitions().contains(300));
            } else {
                Assert.assertTrue(kafkaTaskInfo.getPartitions().contains(200));
            }
        }
    }