@Override
    public boolean addFields(AddFieldRequest addFieldsRequest) {
        String groupId = addFieldsRequest.getInlongGroupId();
        String streamId = addFieldsRequest.getInlongStreamId();
        try {
            LOGGER.info("begin to add inlong stream fields ={}", addFieldsRequest.getSinkFieldList());
            Set<String> existFieldList = streamFieldMapper.selectByIdentifier(groupId, streamId).stream()
                    .map(InlongStreamFieldEntity::getFieldName).collect(Collectors.toSet());
            List<InlongStreamFieldEntity> needAddFieldList = new ArrayList<>();
            for (SinkField sinkField : addFieldsRequest.getSinkFieldList()) {
                if (existFieldList.contains(sinkField.getSourceFieldName())) {
                    LOGGER.info("current stream field={} is exist for groupId={}, streamId={}",
                            sinkField.getSourceFieldName(), groupId, streamId);
                    continue;
                }
                InlongStreamFieldEntity entity = new InlongStreamFieldEntity();
                entity.setFieldName(sinkField.getSourceFieldName());
                entity.setFieldType(sinkField.getSourceFieldType());
                entity.setFieldComment(sinkField.getFieldComment());
                entity.setInlongGroupId(groupId);
                entity.setInlongStreamId(streamId);
                entity.setIsDeleted(InlongConstants.UN_DELETED);
                needAddFieldList.add(entity);
            }
            if (CollectionUtils.isNotEmpty(needAddFieldList)) {
                streamFieldMapper.insertAll(needAddFieldList);
            }
            List<StreamSinkEntity> sinkEntityList = sinkMapper.selectByRelatedId(groupId, streamId);
            for (StreamSinkEntity sink : sinkEntityList) {
                sinkService.addFields(sink, addFieldsRequest.getSinkFieldList());
            }
            LOGGER.debug("success add inlong stream fields={}", needAddFieldList);

        } catch (Exception e) {
            LOGGER.error("add inlong stream fields error for groupId={}, streamId={}", groupId, streamId, e);
            throw new BusinessException(ErrorCodeEnum.INVALID_PARAMETER,
                    String.format("add stream fields error : %s", e.getMessage()));
        }
        return true;
    }
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