private void checkNodeIds(String dataflow) throws Exception {
        JsonNode relations = JsonUtils.parseTree(dataflow).get(InlongConstants.STREAMS)
                .get(0).get(InlongConstants.RELATIONS);
        List<Pair<List<String>, List<String>>> nodeIdsPairList = new ArrayList<>();
        for (int i = 0; i < relations.size(); i++) {
            List<String> inputIds = new ArrayList<>(
                    JsonUtils.OBJECT_MAPPER.convertValue(relations.get(i).get(InlongConstants.INPUTS),
                            new TypeReference<List<String>>() {
                            }));
            if (CollectionUtils.isEmpty(inputIds)) {
                String message = String.format("input nodeId %s cannot be empty", inputIds);
                log.error(message);
                throw new Exception(message);
            }

            List<String> outputIds = new ArrayList<>(
                    JsonUtils.OBJECT_MAPPER.convertValue(relations.get(i).get(InlongConstants.OUTPUTS),
                            new TypeReference<List<String>>() {
                            }));
            if (CollectionUtils.isEmpty(outputIds)) {
                String message = String.format("output nodeId %s cannot be empty", outputIds);
                log.error(message);
                throw new Exception(message);
            }

            if (!Collections.disjoint(inputIds, outputIds)) {
                String message = String.format("input nodeId %s cannot be equal to output nodeId %s",
                        inputIds, outputIds);
                log.error(message);
                throw new Exception(message);
            }
            nodeIdsPairList.add(Pair.of(inputIds, outputIds));
        }

        if (nodeIdsPairList.size() > 1) {
            List<String> allNodeIds = new ArrayList<>(nodeIdsPairList.get(0).getLeft());
            allNodeIds.addAll(nodeIdsPairList.get(0).getRight());
            for (int i = 1; i < relations.size(); i++) {
                if (!Collections.disjoint(allNodeIds, nodeIdsPairList.get(i).getLeft())) {
                    String message = String.format("input nodeId %s already exists ", nodeIdsPairList.get(i).getLeft());
                    log.error(message);
                    throw new Exception(message);
                }
                allNodeIds.addAll(nodeIdsPairList.get(i).getLeft());
            }
        }
    }
Start the Flink job, if the job id was not empty, restore it.
inlong-manager/manager-plugins/base/src/main/java/org/apache/inlong/manager/plugin/flink/FlinkOperation.java