public void removeOldLoadJob() {
        // clean expired load job
        long currentTimeMs = System.currentTimeMillis();

        writeLock();
        try {
            // add load job to a sorted tree set
            Set<LoadJob> jobs = new TreeSet<>(new Comparator<LoadJob>() {
                @Override
                public int compare(LoadJob o1, LoadJob o2) {
                    // sort by finish time desc
                    return Long.signum(o1.getFinishTimestamp() - o2.getFinishTimestamp());
                }
            });

            for (Map.Entry<Long, LoadJob> entry : idToLoadJob.entrySet()) {
                LoadJob job = entry.getValue();
                if (!job.isCompleted()) {
                    continue;
                }
                if (isJobExpired(job, currentTimeMs)) {
                    // remove expired job
                    LOG.info("remove expired job: {}", job.getLabel());
                    unprotectedRemoveJobReleatedMeta(job);
                } else {
                    jobs.add(job);
                }
            }

            // if there are still more jobs than LABEL_KEEP_MAX_NUM
            // remove the ones that finished earlier
            int numJobsToRemove = idToLoadJob.size() - Config.label_keep_max_num;
            if (numJobsToRemove > 0) {
                LOG.info("remove {} jobs from {}", numJobsToRemove, jobs.size());
                Iterator<LoadJob> iterator = jobs.iterator();
                for (int i = 0; i != numJobsToRemove && iterator.hasNext(); ++i) {
                    LoadJob job = iterator.next();
                    LOG.info("remove redundant job: {}", job.getLabel());
                    unprotectedRemoveJobReleatedMeta(job);
                }
            }
        } finally {
            writeUnlock();
        }
    }
@Test
    public void testRemoveOldLoadJob(@Mocked GlobalStateMgr globalStateMgr,
                                     @Injectable Database db) throws Exception {
        new Expectations() {
            {
                globalStateMgr.getDb(anyLong);
                result = db;
            }
        };

        loadManager = new LoadMgr(new LoadJobScheduler());
        int origLabelKeepMaxSecond = Config.label_keep_max_second;
        int origLabelKeepMaxNum = Config.label_keep_max_num;
        Map<Long, LoadJob> idToLoadJob = Deencapsulation.getField(loadManager, "idToLoadJob");
        Map<Long, Map<String, List<LoadJob>>> dbIdToLabelToLoadJobs = Deencapsulation.getField(
                loadManager, "dbIdToLabelToLoadJobs");
        long currentTimeMs = System.currentTimeMillis();

        // finished insert job
        LoadJob job0 = new InsertLoadJob("job0", 0L, 1L, currentTimeMs - 101000, "", "", null);
        job0.id = 10;
        job0.finishTimestamp = currentTimeMs - 101000;
        Deencapsulation.invoke(loadManager, "addLoadJob", job0);

        // broker load job
        // loading
        LoadJob job1 = new BrokerLoadJob(1L, "job1", null, null, null);
        job1.state = JobState.LOADING;
        job1.id = 11;
        Deencapsulation.invoke(loadManager, "addLoadJob", job1);
        // cancelled
        LoadJob job2 = new BrokerLoadJob(1L, "job2", null, null, null);
        job2.finishTimestamp = currentTimeMs - 3000;
        job2.state = JobState.CANCELLED;
        job2.id = 16;
        Deencapsulation.invoke(loadManager, "addLoadJob", job2);
        // finished
        LoadJob job22 = new BrokerLoadJob(1L, "job2", null, null, null);
        job22.finishTimestamp = currentTimeMs - 1000;
        job22.state = JobState.FINISHED;
        job22.id = 12;
        Deencapsulation.invoke(loadManager, "addLoadJob", job22);

        // spark load job
        // etl
        LoadJob job3 = new SparkLoadJob(2L, "job3", null, null);
        job3.state = JobState.ETL;
        job3.id = 13;
        Deencapsulation.invoke(loadManager, "addLoadJob", job3);
        // cancelled
        LoadJob job4 = new SparkLoadJob(2L, "job4", null, null);
        job4.finishTimestamp = currentTimeMs - 51000;
        job4.state = JobState.CANCELLED;
        job4.id = 14;
        Deencapsulation.invoke(loadManager, "addLoadJob", job4);
        // finished
        LoadJob job42 = new SparkLoadJob(2L, "job4", null, null);
        job42.finishTimestamp = currentTimeMs - 2000;
        job42.state = JobState.FINISHED;
        job42.id = 15;
        Deencapsulation.invoke(loadManager, "addLoadJob", job42);

        Assert.assertEquals(7, idToLoadJob.size());
        Assert.assertEquals(3, dbIdToLabelToLoadJobs.size());

        // test remove jobs by label_keep_max_second
        // remove db 0, job0
        Config.label_keep_max_second = 100;
        Config.label_keep_max_num = 10;
        loadManager.removeOldLoadJob();
        System.out.println(idToLoadJob);
        Assert.assertEquals(6, idToLoadJob.size());
        Assert.assertFalse(idToLoadJob.containsKey(10L));
        Assert.assertEquals(2, dbIdToLabelToLoadJobs.size());
        Assert.assertFalse(dbIdToLabelToLoadJobs.containsKey(0L));

        // remove cancelled job4
        Config.label_keep_max_second = 50;
        Config.label_keep_max_num = 10;
        loadManager.removeOldLoadJob();
        System.out.println(idToLoadJob);
        Assert.assertEquals(5, idToLoadJob.size());
        Assert.assertFalse(idToLoadJob.containsKey(14L));
        Assert.assertEquals(2, dbIdToLabelToLoadJobs.size());
        Assert.assertEquals(1, dbIdToLabelToLoadJobs.get(2L).get("job4").size());

        // test remove jobs by label_keep_max_num
        // remove cancelled job2, finished job4
        Config.label_keep_max_second = 50;
        Config.label_keep_max_num = 3;
        loadManager.removeOldLoadJob();
        System.out.println(idToLoadJob);
        Assert.assertEquals(3, idToLoadJob.size());
        Assert.assertFalse(idToLoadJob.containsKey(15L));
        Assert.assertFalse(idToLoadJob.containsKey(16L));
        Assert.assertEquals(2, dbIdToLabelToLoadJobs.size());
        Assert.assertEquals(1, dbIdToLabelToLoadJobs.get(1L).get("job2").size());
        Assert.assertFalse(dbIdToLabelToLoadJobs.get(2L).containsKey("job4"));

        // remove finished job2
        Config.label_keep_max_second = 50;
        Config.label_keep_max_num = 1;
        loadManager.removeOldLoadJob();
        System.out.println(idToLoadJob);
        Assert.assertEquals(2, idToLoadJob.size());
        Assert.assertFalse(idToLoadJob.containsKey(12L));
        Assert.assertEquals(2, dbIdToLabelToLoadJobs.size());
        Assert.assertFalse(dbIdToLabelToLoadJobs.get(1L).containsKey("job2"));

        // recover config
        Config.label_keep_max_second = origLabelKeepMaxSecond;
        Config.label_keep_max_num = origLabelKeepMaxNum;
    }