private GroupInfo getGroupInfo(InlongGroupInfo groupInfo, InlongStreamInfo inlongStreamInfo,
            List<StreamSink> sinkInfos) {
        String streamId = inlongStreamInfo.getInlongStreamId();
        // get source info
        Map<String, List<StreamSource>> sourceMap = sourceService.getSourcesMap(groupInfo,
                Collections.singletonList(inlongStreamInfo));
        List<TransformResponse> transformList = transformService.listTransform(groupInfo.getInlongGroupId(), streamId);
        Map<String, List<TransformResponse>> transformMap = transformList.stream()
                .collect(Collectors.groupingBy(TransformResponse::getInlongStreamId, HashMap::new,
                        Collectors.toCollection(ArrayList::new)));

        List<StreamInfo> sortStreamInfos = new ArrayList<>();
        Map<String, StreamField> fieldMap = new HashMap<>();
        inlongStreamInfo.getSourceList().forEach(
                source -> parseConstantFieldMap(source.getSourceName(), source.getFieldList(), fieldMap));

        List<TransformResponse> transformResponses = transformMap.get(streamId);
        if (CollectionUtils.isNotEmpty(transformResponses)) {
            transformResponses.forEach(
                    trans -> parseConstantFieldMap(trans.getTransformName(), trans.getFieldList(), fieldMap));
        }

        // build a stream info from the nodes and relations
        List<StreamSource> sources = sourceMap.get(streamId);
        for (StreamSink sinkInfo : sinkInfos) {
            CommonBeanUtils.copyProperties(inlongStreamInfo, sinkInfo, true);
            addAuditId(sinkInfo.getProperties(), sinkInfo.getSinkType(), IndicatorType.SEND_SUCCESS);
        }

        for (StreamSource source : sources) {
            source.setFieldList(new ArrayList<>(inlongStreamInfo.getFieldList()));
        }
        List<NodeRelation> relations;

        if (InlongConstants.STANDARD_MODE.equals(groupInfo.getInlongGroupMode())) {
            if (CollectionUtils.isNotEmpty(transformResponses)) {
                relations = NodeRelationUtils.createNodeRelations(inlongStreamInfo);
                // in standard mode(include Data Ingestion and Synchronization), replace upstream source node and
                // transform input fields node to MQ node (which is InLong stream id)
                String mqNodeName = sources.get(0).getSourceName();
                Set<String> nodeNameSet = getInputNodeNames(sources, transformResponses);
                adjustTransformField(transformResponses, nodeNameSet, mqNodeName);
                adjustNodeRelations(relations, nodeNameSet, mqNodeName);
            } else {
                relations = NodeRelationUtils.createNodeRelations(sources, sinkInfos);
            }

            for (int i = 0; i < sources.size(); i++) {
                addAuditId(sources.get(i).getProperties(), sinkInfos.get(0).getSinkType(),
                        IndicatorType.RECEIVED_SUCCESS);
            }
        } else {
            if (CollectionUtils.isNotEmpty(transformResponses)) {
                List<String> sourcesNames = sources.stream().map(StreamSource::getSourceName)
                        .collect(Collectors.toList());
                List<String> transFormNames = transformResponses.stream().map(TransformResponse::getTransformName)
                        .collect(Collectors.toList());
                relations = Arrays.asList(NodeRelationUtils.createNodeRelation(sourcesNames, transFormNames),
                        NodeRelationUtils.createNodeRelation(transFormNames,
                                sinkInfos.stream().map(StreamSink::getSinkName).collect(Collectors.toList())));
            } else {
                relations = NodeRelationUtils.createNodeRelations(sources, sinkInfos);
            }

            for (StreamSource source : sources) {
                addAuditId(source.getProperties(), source.getSourceType(), IndicatorType.RECEIVED_SUCCESS);
            }
        }

        // create extract-transform-load nodes
        List<Node> nodes = this.createNodes(sources, transformResponses, sinkInfos, fieldMap);

        StreamInfo streamInfo = new StreamInfo(streamId, nodes, relations);
        sortStreamInfos.add(streamInfo);

        // rebuild joinerNode relation
        NodeRelationUtils.optimizeNodeRelation(streamInfo, transformResponses);

        return new GroupInfo(groupInfo.getInlongGroupId(), sortStreamInfos);
    }
Default Sort config operator, used to create a Sort config for the InlongGroup with ZK disabled.
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/sort/SortFlinkConfigOperator.java