@Override
    public List<SortStatusInfo> pollSortStatus(List<InlongStreamInfo> streamInfos, String credentials) {
        log.debug("begin to poll sort status for stream");
        if (CollectionUtils.isEmpty(streamInfos)) {
            log.debug("end to poll sort status, as the stream list is empty");
            return Collections.emptyList();
        }

        List<SortStatusInfo> statusInfos = new ArrayList<>(streamInfos.size());
        for (InlongStreamInfo streamInfo : streamInfos) {
            try {
                List<InlongStreamExtInfo> extList = streamInfo.getExtList();
                log.debug("stream {} ext info: {}", streamInfo.getInlongStreamId(), extList);

                Map<String, String> kvConf = new HashMap<>();
                extList.forEach(v -> kvConf.put(v.getKeyName(), v.getKeyValue()));
                String sortExt = kvConf.get(InlongConstants.SORT_PROPERTIES);
                if (StringUtils.isNotEmpty(sortExt)) {
                    Map<String, String> result = JsonUtils.OBJECT_MAPPER.convertValue(
                            JsonUtils.OBJECT_MAPPER.readTree(sortExt), new TypeReference<Map<String, String>>() {
                            });
                    kvConf.putAll(result);
                }

                String jobId = kvConf.get(InlongConstants.SORT_JOB_ID);
                SortStatusInfo statusInfo = SortStatusInfo.builder()
                        .inlongGroupId(streamInfo.getInlongGroupId())
                        .inlongStreamId(streamInfo.getInlongStreamId())
                        .build();
                if (StringUtils.isBlank(jobId)) {
                    statusInfo.setSortStatus(SortStatus.NOT_EXISTS);
                    statusInfos.add(statusInfo);
                    continue;
                }

                String sortUrl = kvConf.get(InlongConstants.SORT_URL);
                statusInfo.setSortStatus(
                        JOB_SORT_STATUS_MAP.getOrDefault(FlinkService.getInstance().getJobStatus(sortUrl, jobId),
                                SortStatus.UNKNOWN));
                statusInfos.add(statusInfo);
            } catch (Exception e) {
                log.error("polling sort status failed for groupId=" + streamInfo.getInlongGroupId() + " streamId="
                        + streamInfo.getInlongStreamId(), e);
            }
        }

        log.debug("success to get sort status: {}", statusInfos);
        return statusInfos;
    }
Flink job status to InLong sort status mapping.
inlong-manager/manager-plugins/base/src/main/java/org/apache/inlong/manager/plugin/poller/SortStatusPoller.java