@Override
    public PageResult<InlongStreamInfo> listAll(InlongStreamPageRequest request) {
        LOGGER.debug("begin to list full inlong stream page by {}", request);
        Preconditions.expectNotNull(request, "request is empty");
        String groupId = request.getInlongGroupId();
        Preconditions.expectNotBlank(groupId, ErrorCodeEnum.GROUP_ID_IS_EMPTY);
        InlongGroupEntity groupEntity = groupMapper.selectByGroupId(groupId);
        Preconditions.expectNotNull(groupEntity, "inlong group not found by groupId=" + groupId);

        // the person in charge of the inlong group has the authority of all inlong streams,
        // so do not filter by in charge person
        PageHelper.startPage(request.getPageNum(), request.getPageSize());
        OrderFieldEnum.checkOrderField(request);
        OrderTypeEnum.checkOrderType(request);
        Page<InlongStreamEntity> page = (Page<InlongStreamEntity>) streamMapper.selectByCondition(request);

        PageResult<InlongStreamInfo> pageResult = PageResult.fromPage(page)
                .map(entity -> CommonBeanUtils.copyProperties(entity, InlongStreamInfo::new))
                .foreach(streamInfo -> {
                    // Set the field information of the inlong stream
                    String streamId = streamInfo.getInlongStreamId();
                    unpackExtParams(streamInfo);
                    List<StreamField> streamFields = getStreamFields(groupId, streamId);
                    streamInfo.setFieldList(streamFields);
                    List<InlongStreamExtEntity> extEntities = streamExtMapper.selectByRelatedId(groupId, streamId);
                    List<InlongStreamExtInfo> streamExtInfos = CommonBeanUtils.copyListProperties(
                            extEntities, InlongStreamExtInfo::new);
                    streamInfo.setExtList(streamExtInfos);

                    // query all valid stream sources
                    List<StreamSource> sourceList = sourceService.listSource(groupId, streamId);
                    streamInfo.setSourceList(sourceList);

                    // query all valid stream sinks and its extended info, field info
                    List<StreamSink> sinkList = sinkService.listSink(groupId, streamId);
                    streamInfo.setSinkList(sinkList);
                });

        LOGGER.debug("success to list full inlong stream info by {}", request);
        return pageResult;
    }
Query and set the extended information and data source fields of the inlong stream
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/stream/InlongStreamServiceImpl.java