public static ListenerResult submitFlinkJob(InlongStreamInfo streamInfo, String jobName,
            boolean isBatchJob, Boundaries boundaries) throws Exception {
        List<StreamSink> sinkList = streamInfo.getSinkList();
        List<String> sinkTypes = sinkList.stream().map(StreamSink::getSinkType).collect(Collectors.toList());
        if (CollectionUtils.isEmpty(sinkList) || !SinkType.containSortFlinkSink(sinkTypes)) {
            log.warn("not any valid sink configured for groupId {} and streamId {}, reason: {},"
                    + " skip launching sort job",
                    CollectionUtils.isEmpty(sinkList) ? "no sink configured" : "no sort flink sink configured",
                    streamInfo.getInlongGroupId(), streamInfo.getInlongStreamId());
            return ListenerResult.success();
        }
        List<InlongStreamExtInfo> extList = streamInfo.getExtList();
        log.info("stream ext info: {}", extList);
        Map<String, String> kvConf = extList.stream().filter(v -> StringUtils.isNotEmpty(v.getKeyName())
                && StringUtils.isNotEmpty(v.getKeyValue())).collect(Collectors.toMap(
                        InlongStreamExtInfo::getKeyName,
                        InlongStreamExtInfo::getKeyValue));

        String sortExtProperties = kvConf.get(InlongConstants.SORT_PROPERTIES);
        if (StringUtils.isNotEmpty(sortExtProperties)) {
            Map<String, String> result = JsonUtils.OBJECT_MAPPER.convertValue(
                    JsonUtils.OBJECT_MAPPER.readTree(sortExtProperties), new TypeReference<Map<String, String>>() {
                    });
            kvConf.putAll(result);
        }

        String dataflow = kvConf.get(InlongConstants.DATAFLOW);
        if (StringUtils.isEmpty(dataflow)) {
            String message = String.format("dataflow is empty for groupId [%s], streamId [%s]",
                    streamInfo.getInlongGroupId(), streamInfo.getInlongStreamId());
            log.error(message);
            return ListenerResult.fail(message);
        }

        FlinkInfo flinkInfo = new FlinkInfo();
        flinkInfo.setJobName(jobName);
        String sortUrl = kvConf.get(InlongConstants.SORT_URL);
        flinkInfo.setEndpoint(sortUrl);
        flinkInfo.setInlongStreamInfoList(Collections.singletonList(streamInfo));
        if (isBatchJob) {
            flinkInfo.setRuntimeExecutionMode(RUNTIME_EXECUTION_MODE_BATCH);
            flinkInfo.setBoundaryType(boundaries.getBoundaryType().getType());
            flinkInfo.setLowerBoundary(boundaries.getLowerBound());
            flinkInfo.setUpperBoundary(boundaries.getUpperBound());
        } else {
            flinkInfo.setRuntimeExecutionMode(RUNTIME_EXECUTION_MODE_STREAM);
        }
        FlinkOperation flinkOperation = FlinkOperation.getInstance();
        try {
            flinkOperation.genPath(flinkInfo, dataflow);
            flinkOperation.start(flinkInfo);
            log.info("job submit success for groupId = {}, streamId = {}, jobId = {}",
                    streamInfo.getInlongGroupId(), streamInfo.getInlongStreamId(), flinkInfo.getJobId());
        } catch (Exception e) {
            flinkInfo.setException(true);
            flinkInfo.setExceptionMsg(getExceptionStackMsg(e));
            flinkOperation.pollJobStatus(flinkInfo, JobStatus.RUNNING);

            String message = String.format("startup sort failed for groupId [%s], streamId [%s]",
                    streamInfo.getInlongGroupId(), streamInfo.getInlongStreamId());
            log.error(message, e);
            return ListenerResult.fail(message + e.getMessage());
        }

        saveInfo(streamInfo, InlongConstants.SORT_JOB_ID, flinkInfo.getJobId(), extList);
        flinkOperation.pollJobStatus(flinkInfo, JobStatus.RUNNING);
        return ListenerResult.success();
    }
writeConfigToFile

@param configJobDirectory job config directory
@param configFileName config file name
@param content contents of the file to be written
@return whether success
inlong-manager/manager-plugins/base/src/main/java/org/apache/inlong/manager/plugin/util/FlinkUtils.java