public static ListenerResult submitFlinkJobs(String groupId, List<InlongStreamInfo> streamInfoList,
            boolean isBatchJob, Boundaries boundaries) throws Exception {
        int sinkCount = streamInfoList.stream()
                .map(s -> s.getSinkList() == null ? 0 : s.getSinkList().size())
                .reduce(0, Integer::sum);
        if (sinkCount == 0) {
            log.warn("Not any sink configured for group {} and stream list {}, skip launching sort job", groupId,
                    streamInfoList.stream()
                            .map(s -> s.getInlongGroupId() + ":" + s.getName()).collect(Collectors.toList()));
            return ListenerResult.success();
        }

        List<ListenerResult> listenerResults = new ArrayList<>();
        for (InlongStreamInfo streamInfo : streamInfoList) {
            listenerResults.add(
                    FlinkUtils.submitFlinkJob(
                            streamInfo, FlinkUtils.genFlinkJobName(streamInfo), isBatchJob, boundaries));
        }

        // only one stream in group for now
        // we can return the list of ListenerResult if support multi-stream in the future
        List<ListenerResult> failedStreams = listenerResults.stream()
                .filter(t -> !t.isSuccess()).collect(Collectors.toList());
        if (failedStreams.isEmpty()) {
            return ListenerResult.success();
        }
        return ListenerResult.fail(failedStreams.get(0).getRemark());
    }
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