private Configuration getFlinkConfiguration(String endpoint) {
        return configurations.computeIfAbsent(endpoint,
                k -> {
                    Integer port;
                    String address;
                    if (StringUtils.isEmpty(endpoint)) {
                        address = flinkConfig.getAddress();
                        port = flinkConfig.getPort();
                    } else {
                        Map<String, String> ipPort = translateFromEndpoint(endpoint);
                        if (ipPort.isEmpty()) {
                            throw new BusinessException("get address:port failed from endpoint " + endpoint);
                        }
                        address = ipPort.get("address");
                        port = Integer.valueOf(ipPort.get("port"));
                    }
                    // build flink configuration
                    Configuration configuration = new Configuration();
                    configuration.setInteger(JobManagerOptions.PORT, flinkConfig.getJobManagerPort());
                    configuration.setString(JobManagerOptions.ADDRESS, address);
                    configuration.setInteger(RestOptions.PORT, port);
                    return configuration;
                });
    }
Get the job status by the given job id.
inlong-manager/manager-plugins/base/src/main/java/org/apache/inlong/manager/plugin/flink/FlinkService.java