[Feature][Core] Add PendingJobs info in the response of GetOverviewOperation

[xiaochen-zhou]

Purpose of this pull request

Add PendingJobs info in the response of GetOverviewOperation

Does this PR introduce any user-facing change?

How was this patch tested?

Check list

• If any new Jar binary package adding in your PR, please add License Notice according
  New License Guide
• If necessary, please update the documentation to describe the new feature. https://github.com/apache/seatunnel/tree/dev/docs
• If you are contributing the connector code, please check that the following files are updated:
  1. Update plugin-mapping.properties and add new connector information in it
  2. Update the pom file of seatunnel-dist
  3. Add ci label in label-scope-conf
  4. Add e2e testcase in seatunnel-e2e
  5. Update connector plugin_config

[dybyte]

What do you think about creating a method in CoordinatorService that directly returns only the pendingJobCount instead of using the getJobCountMetrics method?

[Hisoka-X]

Let's use

seatunnel/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/utils/PeekBlockingQueue.java

Line 51 in b5f583b

private final Map<Long, E> jobIdMap = new ConcurrentHashMap<>();

[hawk9821]

• the constructor method of the JobHistoryService

jobHistoryService =
                new JobHistoryService(
                        nodeEngine,
                        runningJobStateIMap,
                        logger,
                        pendingJobQueue.getJobIdMap(),
                        runningJobMasterMap,
                        nodeEngine.getHazelcastInstance().getMap(Constant.IMAP_FINISHED_JOB_STATE),
                        nodeEngine
                                .getHazelcastInstance()
                                .getMap(Constant.IMAP_FINISHED_JOB_METRICS),
                        nodeEngine
                                .getHazelcastInstance()
                                .getMap(Constant.IMAP_FINISHED_JOB_VERTEX_INFO),
                        engineConfig.getHistoryJobExpireMinutes());

• the getJobStatusData method of JobHistoryService

public List<JobStatusData> getJobStatusData() {
        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());

        List<JobState> pendingJobStateList =
                pendingJobInfoMap.entrySet().stream()
                        .map(
                                entry -> {
                                    Long jobId = entry.getKey();
                                    JobImmutableInformation jobImmutableInformation =
                                            entry.getValue()
                                                    .getJobMaster()
                                                    .getJobImmutableInformation();
                                    return new JobState(
                                            jobId,
                                            jobImmutableInformation.getJobName(),
                                            JobStatus.PENDING,
                                            jobImmutableInformation.getCreateTime(),
                                            null,
                                            null,
                                            null,
                                            null);
                                })
                        .collect(Collectors.toList());
....

jobIdMap has been assigned to pendingJobInfoMap of the JobHistoryService

[dybyte]

In my opinion, the getJobStatusData() and getJobCountMetrics() methods perform a lot of extra work that isn’t needed if we only want to get the pendingJobCount.