public String listAllJob() {
        List<JobStatusData> status = new ArrayList<>();
        final List<JobState> runningJobStateList =
                runningJobMasterMap.values().stream()
                        .map(master -> toJobStateMapper(master, true))
                        .collect(Collectors.toList());
        Set<Long> runningJonIds =
                runningJobStateList.stream().map(JobState::getJobId).collect(Collectors.toSet());
        Stream.concat(
                        runningJobStateList.stream(),
                        finishedJobStateImap.values().stream()
                                .filter(jobState -> !runningJonIds.contains(jobState.getJobId())))
                .forEach(
                        jobState -> {
                            JobStatusData jobStatusData =
                                    new JobStatusData(
                                            jobState.getJobId(),
                                            jobState.getJobName(),
                                            jobState.getJobStatus(),
                                            jobState.getSubmitTime(),
                                            jobState.getFinishTime());
                            status.add(jobStatusData);
                        });
        try {
            return objectMapper.writeValueAsString(status);
        } catch (JsonProcessingException e) {
            logger.severe("Failed to list all job", e);
            throw new SeaTunnelEngineException(e);
        }
    }
@Test
    public void testlistJobState() throws Exception {
        startJob(JOB_1, "fake_to_console.conf");

        // waiting for JOB_1 status turn to RUNNING
        await().atMost(60000, TimeUnit.MILLISECONDS)
                .untilAsserted(
                        () ->
                                Assertions.assertTrue(
                                        server.getCoordinatorService()
                                                .getJobHistoryService()
                                                .listAllJob()
                                                .contains(
                                                        String.format(
                                                                "\"jobId\":%s,\"jobName\":\"Test\",\"jobStatus\":\"RUNNING\"",
                                                                JOB_1))));

        // waiting for JOB_1 status turn to FINISHED
        await().atMost(60000, TimeUnit.MILLISECONDS)
                .untilAsserted(
                        () ->
                                Assertions.assertTrue(
                                        server.getCoordinatorService()
                                                .getJobHistoryService()
                                                .listAllJob()
                                                .contains(
                                                        String.format(
                                                                "\"jobId\":%s,\"jobName\":\"Test\",\"jobStatus\":\"FINISHED\"",
                                                                JOB_1))));

        startJob(JOB_2, "fake_to_console.conf");
        // waiting for JOB_2 status turn to FINISHED and JOB_2 status turn to RUNNING
        await().atMost(60000, TimeUnit.MILLISECONDS)
                .untilAsserted(
                        () ->
                                Assertions.assertTrue(
                                        server.getCoordinatorService()
                                                        .getJobHistoryService()
                                                        .listAllJob()
                                                        .contains(
                                                                String.format(
                                                                        "\"jobId\":%s,\"jobName\":\"Test\",\"jobStatus\":\"FINISHED\"",
                                                                        JOB_1))
                                                && server.getCoordinatorService()
                                                        .getJobHistoryService()
                                                        .listAllJob()
                                                        .contains(
                                                                String.format(
                                                                        "\"jobId\":%s,\"jobName\":\"Test\",\"jobStatus\":\"RUNNING\"",
                                                                        JOB_2))));
    }