public static void init(long intervalMs) {
        checkers.put(JobState.PENDING, new ExportChecker(JobState.PENDING, intervalMs));
        checkers.put(JobState.EXPORTING, new ExportChecker(JobState.EXPORTING, intervalMs));

        int poolSize = Config.export_running_job_num_limit == 0 ? 5 : Config.export_running_job_num_limit;
        LeaderTaskExecutor pendingTaskExecutor = new LeaderTaskExecutor("export_pending_job", poolSize, true);
        executors.put(JobState.PENDING, pendingTaskExecutor);

        LeaderTaskExecutor exportingTaskExecutor = new LeaderTaskExecutor("export_exporting_job", poolSize, true);
        executors.put(JobState.EXPORTING, exportingTaskExecutor);

        // One export job will be split into multiple exporting sub tasks, the queue size is not determined, so set Integer.MAX_VALUE.
        exportingSubTaskExecutor = new LeaderTaskExecutor("export_exporting_sub_task", Config.export_task_pool_size,
                Integer.MAX_VALUE, true);
    }
@Test
    public void testCheckBeStatus() throws NoSuchFieldException, SecurityException, 
        IllegalArgumentException, IllegalAccessException, NoSuchMethodException, InvocationTargetException {

        new MockUp<ExportJob>() {
            @Mock
            public synchronized void cancel(ExportFailMsg.CancelType type, String msg) throws UserException {
            }
        };

        Backend be = new Backend();
        
        new MockUp<SystemInfoService>() {
            @Mock
            public ComputeNode getBackendOrComputeNode(long backendId) {
                return be;
            }
        };

        ExportChecker.init(1000L);
        Field field = ExportChecker.class.getDeclaredField("checkers");
        field.setAccessible(true);
        Object obj = field.get(ExportChecker.class);
        
        Map<JobState, ExportChecker> map = (Map<JobState, ExportChecker>) obj;
        ExportChecker checker = map.get(JobState.EXPORTING);
        Method method = ExportChecker.class.getDeclaredMethod("checkJobNeedCancel", ExportJob.class);
        method.setAccessible(true);

        ExportJob job = new ExportJob();
        job.setBeStartTime(1, 1000L);
        boolean cancelled = (boolean) method.invoke(checker, job);
        Assert.assertTrue(cancelled);

        be.setAlive(true);
        be.setDecommissioned(true);

        be.setLastStartTime(1001L);

        cancelled = (boolean) method.invoke(checker, job);
        Assert.assertTrue(cancelled);

        be.setLastStartTime(999L);

        cancelled = (boolean) method.invoke(checker, job);
        Assert.assertTrue(!cancelled);

        new MockUp<SystemInfoService>() {
            @Mock
            public ComputeNode getBackendOrComputeNode(long backendId) {
                return null;
            }
        };

        cancelled = (boolean) method.invoke(checker, job);
        Assert.assertTrue(cancelled);
    }