This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 8c9d7c3fc [INLONG-8039][Manager] Optimize the transform interface to
support pagination query (#8040)
8c9d7c3fc is described below
commit 8c9d7c3fc6d2971c0e32d9a49e6ce38db98f866e
Author: haifxu <[email protected]>
AuthorDate: Thu May 18 14:38:03 2023 +0800
[INLONG-8039][Manager] Optimize the transform interface to support
pagination query (#8040)
---
.../dao/mapper/StreamTransformEntityMapper.java | 3 +
.../mappers/StreamTransformEntityMapper.xml | 21 +++
.../pojo/transform/TransformPageRequest.java | 45 ++++++
.../resource/sort/DefaultSortConfigOperator.java | 2 -
.../service/transform/StreamTransformService.java | 19 +++
.../transform/StreamTransformServiceImpl.java | 167 ++++++++++++---------
.../transform/StreamTransformServiceTest.java | 4 +-
.../web/controller/StreamTransformController.java | 23 ++-
8 files changed, 201 insertions(+), 83 deletions(-)
diff --git
a/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/mapper/StreamTransformEntityMapper.java
b/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/mapper/StreamTransformEntityMapper.java
index a087624a0..129c51b36 100644
---
a/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/mapper/StreamTransformEntityMapper.java
+++
b/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/mapper/StreamTransformEntityMapper.java
@@ -19,6 +19,7 @@ package org.apache.inlong.manager.dao.mapper;
import org.apache.ibatis.annotations.Param;
import org.apache.inlong.manager.dao.entity.StreamTransformEntity;
+import org.apache.inlong.manager.pojo.transform.TransformPageRequest;
import org.springframework.stereotype.Repository;
import java.util.List;
@@ -35,6 +36,8 @@ public interface StreamTransformEntityMapper {
List<StreamTransformEntity> selectByRelatedId(@Param("groupId") String
groupId, @Param("streamId") String streamId,
@Param("transformName") String transformName);
+ List<StreamTransformEntity> selectByCondition(@Param("request")
TransformPageRequest request);
+
int updateById(StreamTransformEntity record);
int updateByIdSelective(StreamTransformEntity record);
diff --git
a/inlong-manager/manager-dao/src/main/resources/mappers/StreamTransformEntityMapper.xml
b/inlong-manager/manager-dao/src/main/resources/mappers/StreamTransformEntityMapper.xml
index 302e5f81f..9ca419fc0 100644
---
a/inlong-manager/manager-dao/src/main/resources/mappers/StreamTransformEntityMapper.xml
+++
b/inlong-manager/manager-dao/src/main/resources/mappers/StreamTransformEntityMapper.xml
@@ -142,6 +142,27 @@
</where>
</select>
+ <select id="selectByCondition"
+
parameterType="org.apache.inlong.manager.pojo.transform.TransformPageRequest"
+
resultType="org.apache.inlong.manager.dao.entity.StreamTransformEntity">
+ select
+ <include refid="Base_Column_List"/>
+ from stream_transform
+ <where>
+ is_deleted = 0
+ and inlong_group_id = #{request.inlongGroupId, jdbcType=VARCHAR}
+ <if test="request.inlongStreamId != null and
request.inlongStreamId != ''">
+ and inlong_stream_id = #{request.inlongStreamId,
jdbcType=VARCHAR}
+ </if>
+ <if test="request.transformType != null and request.transformType
!= ''">
+ and transform_type = #{request.transformType, jdbcType=VARCHAR}
+ </if>
+ <if test="request.transformName != null and request.transformName
!= ''">
+ and transform_name = #{request.transformName, jdbcType=VARCHAR}
+ </if>
+ </where>
+ </select>
+
<update id="updateById"
parameterType="org.apache.inlong.manager.dao.entity.StreamTransformEntity">
update stream_transform
set inlong_group_id = #{inlongGroupId,jdbcType=VARCHAR},
diff --git
a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/transform/TransformPageRequest.java
b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/transform/TransformPageRequest.java
new file mode 100644
index 000000000..e33ba5173
--- /dev/null
+++
b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/transform/TransformPageRequest.java
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.manager.pojo.transform;
+
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import org.apache.inlong.manager.pojo.common.PageRequest;
+
+import javax.validation.constraints.NotBlank;
+
+@Data
+@EqualsAndHashCode(callSuper = false)
+@ApiModel("Paging query request for TransformPageRequest")
+public class TransformPageRequest extends PageRequest {
+
+ @NotBlank(message = "inlongGroupId cannot be blank")
+ @ApiModelProperty(value = "Inlong group id", required = true)
+ private String inlongGroupId;
+
+ @ApiModelProperty(value = "Inlong stream id")
+ private String inlongStreamId;
+
+ @ApiModelProperty("Transform name, unique in one stream")
+ private String transformName;
+
+ @ApiModelProperty("Transform type, including: splitter, filter, joiner,
etc.")
+ private String transformType;
+}
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/sort/DefaultSortConfigOperator.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/sort/DefaultSortConfigOperator.java
index e6c56d79f..2f9d16082 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/sort/DefaultSortConfigOperator.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/sort/DefaultSortConfigOperator.java
@@ -102,7 +102,6 @@ public class DefaultSortConfigOperator implements
SortConfigOperator {
Map<String, List<StreamSource>> sourceMap =
sourceService.getSourcesMap(groupInfo, streamInfoList);
// get sink info
Map<String, List<StreamSink>> sinkMap =
sinkService.getSinksMap(groupInfo, streamInfoList);
-
List<TransformResponse> transformList =
transformService.listTransform(groupInfo.getInlongGroupId(), null);
Map<String, List<TransformResponse>> transformMap =
transformList.stream()
.collect(Collectors.groupingBy(TransformResponse::getInlongStreamId,
HashMap::new,
@@ -130,7 +129,6 @@ public class DefaultSortConfigOperator implements
SortConfigOperator {
auditIds.add(auditService.getAuditId(sink.getSinkType(),
false));
}
for (StreamSource source : sources) {
- source.setFieldList(inlongStream.getFieldList());
Map<String, Object> properties = source.getProperties();
properties.putIfAbsent("metrics.audit.key", String.join("&",
auditIds));
}
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/transform/StreamTransformService.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/transform/StreamTransformService.java
index 012aad9dd..f172219c7 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/transform/StreamTransformService.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/transform/StreamTransformService.java
@@ -17,7 +17,9 @@
package org.apache.inlong.manager.service.transform;
+import org.apache.inlong.manager.pojo.common.PageResult;
import org.apache.inlong.manager.pojo.transform.DeleteTransformRequest;
+import org.apache.inlong.manager.pojo.transform.TransformPageRequest;
import org.apache.inlong.manager.pojo.transform.TransformRequest;
import org.apache.inlong.manager.pojo.transform.TransformResponse;
import org.apache.inlong.manager.pojo.user.UserInfo;
@@ -47,6 +49,23 @@ public interface StreamTransformService {
*/
Integer save(TransformRequest request, UserInfo opInfo);
+ /**
+ * Query transform information based on inlong group id and inlong stream
id.
+ *
+ * @param request the transform page request
+ * @return the transform response
+ */
+ PageResult<TransformResponse> listByCondition(TransformPageRequest
request, UserInfo opInfo);
+
+ /**
+ * Query transform information based on id
+ *
+ * @param id transform id.
+ * @param opInfo userinfo of operator
+ * @return transform info
+ */
+ TransformResponse get(Integer id, UserInfo opInfo);
+
/**
* Query transform information based on inlong group id and inlong stream
id.
*
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/transform/StreamTransformServiceImpl.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/transform/StreamTransformServiceImpl.java
index 8d9d33669..fcc104109 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/transform/StreamTransformServiceImpl.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/transform/StreamTransformServiceImpl.java
@@ -17,28 +17,31 @@
package org.apache.inlong.manager.service.transform;
+import com.github.pagehelper.PageHelper;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.inlong.manager.common.consts.InlongConstants;
import org.apache.inlong.manager.common.enums.ErrorCodeEnum;
import org.apache.inlong.manager.common.enums.GroupStatus;
-import org.apache.inlong.manager.common.enums.UserTypeEnum;
import org.apache.inlong.manager.common.exceptions.BusinessException;
-import org.apache.inlong.manager.dao.entity.InlongGroupEntity;
-import org.apache.inlong.manager.dao.mapper.InlongGroupEntityMapper;
-import org.apache.inlong.manager.pojo.stream.StreamField;
-import org.apache.inlong.manager.pojo.transform.DeleteTransformRequest;
-import org.apache.inlong.manager.pojo.transform.TransformRequest;
-import org.apache.inlong.manager.pojo.transform.TransformResponse;
import org.apache.inlong.manager.common.util.CommonBeanUtils;
import org.apache.inlong.manager.common.util.Preconditions;
+import org.apache.inlong.manager.dao.entity.InlongGroupEntity;
import org.apache.inlong.manager.dao.entity.StreamTransformEntity;
import org.apache.inlong.manager.dao.entity.StreamTransformFieldEntity;
+import org.apache.inlong.manager.dao.mapper.InlongGroupEntityMapper;
import org.apache.inlong.manager.dao.mapper.StreamTransformEntityMapper;
import org.apache.inlong.manager.dao.mapper.StreamTransformFieldEntityMapper;
+import org.apache.inlong.manager.pojo.common.PageResult;
+import org.apache.inlong.manager.pojo.stream.StreamField;
+import org.apache.inlong.manager.pojo.transform.DeleteTransformRequest;
+import org.apache.inlong.manager.pojo.transform.TransformPageRequest;
+import org.apache.inlong.manager.pojo.transform.TransformRequest;
+import org.apache.inlong.manager.pojo.transform.TransformResponse;
import org.apache.inlong.manager.pojo.user.UserInfo;
import org.apache.inlong.manager.service.group.GroupCheckService;
+import org.apache.inlong.manager.service.user.UserService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@@ -47,7 +50,6 @@ import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -69,6 +71,8 @@ public class StreamTransformServiceImpl implements
StreamTransformService {
protected StreamTransformFieldEntityMapper transformFieldMapper;
@Autowired
protected GroupCheckService groupCheckService;
+ @Autowired
+ protected UserService userService;
@Override
@Transactional(rollbackFor = Throwable.class, propagation =
Propagation.REQUIRES_NEW)
@@ -107,12 +111,8 @@ public class StreamTransformServiceImpl implements
StreamTransformService {
String.format("InlongGroup does not exist with
InlongGroupId=%s", request.getInlongGroupId()));
}
// only the person in charges can query
- if (!opInfo.getAccountType().equals(UserTypeEnum.ADMIN.getCode())) {
- List<String> inCharges =
Arrays.asList(groupEntity.getInCharges().split(InlongConstants.COMMA));
- if (!inCharges.contains(opInfo.getName())) {
- throw new
BusinessException(ErrorCodeEnum.GROUP_PERMISSION_DENIED);
- }
- }
+ userService.checkUser(groupEntity.getInCharges(), opInfo.getName(),
+ ErrorCodeEnum.GROUP_PERMISSION_DENIED.getMessage());
// check inlong group status
GroupStatus status = GroupStatus.forCode(groupEntity.getStatus());
if (GroupStatus.notAllowedUpdate(status)) {
@@ -139,16 +139,48 @@ public class StreamTransformServiceImpl implements
StreamTransformService {
}
@Override
- public List<TransformResponse> listTransform(String groupId, String
streamId) {
+ public PageResult<TransformResponse> listByCondition(TransformPageRequest
request, UserInfo opInfo) {
+ String groupId = request.getInlongGroupId();
+ String streamId = request.getInlongStreamId();
LOGGER.debug("begin to fetch transform info by groupId={} and
streamId={} ", groupId, streamId);
- Preconditions.expectNotBlank(groupId, ErrorCodeEnum.GROUP_ID_IS_EMPTY);
- List<StreamTransformEntity> entityList =
transformMapper.selectByRelatedId(groupId, streamId, null);
+ PageHelper.startPage(request.getPageNum(), request.getPageSize());
+ PageResult<TransformResponse> pageResponse = new PageResult<>();
+ InlongGroupEntity groupEntity = groupMapper.selectByGroupId(groupId);
+ if (groupEntity == null) {
+ throw new BusinessException(ErrorCodeEnum.GROUP_NOT_FOUND,
+ String.format("InlongGroup does not exist with
InlongGroupId=%s", groupId));
+ }
+ userService.checkUser(groupEntity.getInCharges(), opInfo.getName(),
+ ErrorCodeEnum.GROUP_PERMISSION_DENIED.getMessage());
+
+ // query result
+ List<StreamTransformEntity> entityList =
transformMapper.selectByCondition(request);
if (CollectionUtils.isEmpty(entityList)) {
- return Collections.emptyList();
+ pageResponse.setList(Collections.emptyList());
+ return pageResponse;
}
- List<Integer> transformIds =
entityList.stream().map(StreamTransformEntity::getId).collect(Collectors.toList());
- List<StreamTransformFieldEntity> fieldEntities =
transformFieldMapper.selectByTransformIds(transformIds);
+ pageResponse.setList(getTransformResponse(entityList));
+ return pageResponse;
+ }
+
+ @Override
+ public TransformResponse get(Integer id, UserInfo opInfo) {
+ StreamTransformEntity entity = transformMapper.selectById(id);
+ List<StreamTransformFieldEntity> fieldEntities =
transformFieldMapper.selectByTransformId(id);
+ if (entity == null) {
+ throw new BusinessException(ErrorCodeEnum.TRANSFORM_NOT_FOUND,
+ String.format("transform not found by id=%s", id));
+ }
+
+ InlongGroupEntity groupEntity =
groupMapper.selectByGroupId(entity.getInlongGroupId());
+ if (groupEntity == null) {
+ throw new BusinessException(ErrorCodeEnum.GROUP_NOT_FOUND,
+ String.format("InlongGroup does not exist with
InlongGroupId=%s", entity.getInlongGroupId()));
+ }
+ userService.checkUser(groupEntity.getInCharges(), opInfo.getName(),
+ ErrorCodeEnum.GROUP_PERMISSION_DENIED.getMessage());
+
Map<Integer, List<StreamField>> fieldInfoMap = fieldEntities.stream()
.map(transformFieldEntity -> {
StreamField fieldInfo =
CommonBeanUtils.copyProperties(transformFieldEntity, StreamField::new);
@@ -157,17 +189,19 @@ public class StreamTransformServiceImpl implements
StreamTransformService {
return Pair.of(transformFieldEntity.getTransformId(),
fieldInfo);
}).collect(Collectors.groupingBy(Pair::getLeft,
Collectors.mapping(Pair::getRight,
Collectors.toList())));
- List<TransformResponse> transformResponses = entityList.stream()
- .map(entity -> CommonBeanUtils.copyProperties(entity,
TransformResponse::new))
- .collect(Collectors.toList());
- transformResponses.forEach(transformResponse -> {
- int transformId = transformResponse.getId();
- List<StreamField> fieldInfos = fieldInfoMap.get(transformId);
- if (CollectionUtils.isNotEmpty(fieldInfos)) {
- transformResponse.setFieldList(fieldInfos);
- }
- });
- return transformResponses;
+ TransformResponse transformResponse =
CommonBeanUtils.copyProperties(entity, TransformResponse::new);
+ transformResponse.setFieldList(fieldInfoMap.get(id));
+ return transformResponse;
+ }
+
+ @Override
+ public List<TransformResponse> listTransform(String groupId, String
streamId) {
+ Preconditions.expectNotBlank(groupId, ErrorCodeEnum.GROUP_ID_IS_EMPTY);
+ List<StreamTransformEntity> entityList =
transformMapper.selectByRelatedId(groupId, streamId, null);
+ if (CollectionUtils.isEmpty(entityList)) {
+ return Collections.emptyList();
+ }
+ return getTransformResponse(entityList);
}
@Override
@@ -178,40 +212,15 @@ public class StreamTransformServiceImpl implements
StreamTransformService {
throw new BusinessException(ErrorCodeEnum.GROUP_NOT_FOUND,
String.format("InlongGroup does not exist with
InlongGroupId=%s", groupId));
}
- // only the person in charges can query
- if (!opInfo.getAccountType().equals(UserTypeEnum.ADMIN.getCode())) {
- List<String> inCharges =
Arrays.asList(groupEntity.getInCharges().split(InlongConstants.COMMA));
- if (!inCharges.contains(opInfo.getName())) {
- throw new
BusinessException(ErrorCodeEnum.GROUP_PERMISSION_DENIED);
- }
- }
+ userService.checkUser(groupEntity.getInCharges(), opInfo.getName(),
+ ErrorCodeEnum.GROUP_PERMISSION_DENIED.getMessage());
// query result
List<StreamTransformEntity> entityList =
transformMapper.selectByRelatedId(groupId, streamId, null);
if (CollectionUtils.isEmpty(entityList)) {
return Collections.emptyList();
}
// get transform data
- List<Integer> transformIds =
entityList.stream().map(StreamTransformEntity::getId).collect(Collectors.toList());
- List<StreamTransformFieldEntity> fieldEntities =
transformFieldMapper.selectByTransformIds(transformIds);
- Map<Integer, List<StreamField>> fieldInfoMap = fieldEntities.stream()
- .map(transformFieldEntity -> {
- StreamField fieldInfo =
CommonBeanUtils.copyProperties(transformFieldEntity, StreamField::new);
-
fieldInfo.setFieldType(transformFieldEntity.getFieldType());
- fieldInfo.setId(transformFieldEntity.getRankNum());
- return Pair.of(transformFieldEntity.getTransformId(),
fieldInfo);
- }).collect(Collectors.groupingBy(Pair::getLeft,
- Collectors.mapping(Pair::getRight,
Collectors.toList())));
- List<TransformResponse> transformResponses = entityList.stream()
- .map(entity -> CommonBeanUtils.copyProperties(entity,
TransformResponse::new))
- .collect(Collectors.toList());
- transformResponses.forEach(transformResponse -> {
- int transformId = transformResponse.getId();
- List<StreamField> fieldInfos = fieldInfoMap.get(transformId);
- if (CollectionUtils.isNotEmpty(fieldInfos)) {
- transformResponse.setFieldList(fieldInfos);
- }
- });
- return transformResponses;
+ return getTransformResponse(entityList);
}
@Override
@@ -252,12 +261,8 @@ public class StreamTransformServiceImpl implements
StreamTransformService {
String.format("InlongGroup does not exist with
InlongGroupId=%s", request.getInlongGroupId()));
}
// only the person in charges can query
- if (!opInfo.getAccountType().equals(UserTypeEnum.ADMIN.getCode())) {
- List<String> inCharges =
Arrays.asList(groupEntity.getInCharges().split(InlongConstants.COMMA));
- if (!inCharges.contains(opInfo.getName())) {
- throw new
BusinessException(ErrorCodeEnum.GROUP_PERMISSION_DENIED);
- }
- }
+ userService.checkUser(groupEntity.getInCharges(), opInfo.getName(),
+ ErrorCodeEnum.GROUP_PERMISSION_DENIED.getMessage());
// check inlong group status
GroupStatus status = GroupStatus.forCode(groupEntity.getStatus());
if (GroupStatus.notAllowedUpdate(status)) {
@@ -322,12 +327,8 @@ public class StreamTransformServiceImpl implements
StreamTransformService {
String.format("InlongGroup does not exist with
InlongGroupId=%s", request.getInlongGroupId()));
}
// only the person in charges can query
- if (!opInfo.getAccountType().equals(UserTypeEnum.ADMIN.getCode())) {
- List<String> inCharges =
Arrays.asList(groupEntity.getInCharges().split(InlongConstants.COMMA));
- if (!inCharges.contains(opInfo.getName())) {
- throw new
BusinessException(ErrorCodeEnum.GROUP_PERMISSION_DENIED);
- }
- }
+ userService.checkUser(groupEntity.getInCharges(), opInfo.getName(),
+ ErrorCodeEnum.GROUP_PERMISSION_DENIED.getMessage());
// check inlong group status
GroupStatus status = GroupStatus.forCode(groupEntity.getStatus());
if (GroupStatus.notAllowedUpdate(status)) {
@@ -357,6 +358,30 @@ public class StreamTransformServiceImpl implements
StreamTransformService {
return true;
}
+ private List<TransformResponse>
getTransformResponse(List<StreamTransformEntity> entityList) {
+ List<Integer> transformIds =
entityList.stream().map(StreamTransformEntity::getId).collect(Collectors.toList());
+ List<StreamTransformFieldEntity> fieldEntities =
transformFieldMapper.selectByTransformIds(transformIds);
+ Map<Integer, List<StreamField>> fieldInfoMap = fieldEntities.stream()
+ .map(transformFieldEntity -> {
+ StreamField fieldInfo =
CommonBeanUtils.copyProperties(transformFieldEntity, StreamField::new);
+
fieldInfo.setFieldType(transformFieldEntity.getFieldType());
+ fieldInfo.setId(transformFieldEntity.getRankNum());
+ return Pair.of(transformFieldEntity.getTransformId(),
fieldInfo);
+ }).collect(Collectors.groupingBy(Pair::getLeft,
+ Collectors.mapping(Pair::getRight,
Collectors.toList())));
+ List<TransformResponse> transformResponses = entityList.stream()
+ .map(entity -> CommonBeanUtils.copyProperties(entity,
TransformResponse::new))
+ .collect(Collectors.toList());
+ transformResponses.forEach(transformResponse -> {
+ int transformId = transformResponse.getId();
+ List<StreamField> fieldInfos = fieldInfoMap.get(transformId);
+ if (CollectionUtils.isNotEmpty(fieldInfos)) {
+ transformResponse.setFieldList(fieldInfos);
+ }
+ });
+ return transformResponses;
+ }
+
private void checkParams(TransformRequest request) {
Preconditions.expectNotNull(request,
ErrorCodeEnum.REQUEST_IS_EMPTY.getMessage());
String groupId = request.getInlongGroupId();
diff --git
a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/transform/StreamTransformServiceTest.java
b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/transform/StreamTransformServiceTest.java
index 112123536..7c8e3f830 100644
---
a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/transform/StreamTransformServiceTest.java
+++
b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/transform/StreamTransformServiceTest.java
@@ -19,11 +19,11 @@ package org.apache.inlong.manager.service.transform;
import org.apache.inlong.manager.common.consts.InlongConstants;
import org.apache.inlong.manager.common.enums.TransformType;
-import org.apache.inlong.manager.pojo.transform.TransformRequest;
-import org.apache.inlong.manager.pojo.transform.TransformResponse;
import org.apache.inlong.manager.common.util.CommonBeanUtils;
import org.apache.inlong.manager.dao.entity.StreamTransformEntity;
import org.apache.inlong.manager.dao.mapper.StreamTransformEntityMapper;
+import org.apache.inlong.manager.pojo.transform.TransformRequest;
+import org.apache.inlong.manager.pojo.transform.TransformResponse;
import org.apache.inlong.manager.service.ServiceBaseTest;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
diff --git
a/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/StreamTransformController.java
b/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/StreamTransformController.java
index ff00e1149..6cd77f660 100644
---
a/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/StreamTransformController.java
+++
b/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/StreamTransformController.java
@@ -18,11 +18,14 @@
package org.apache.inlong.manager.web.controller;
import io.swagger.annotations.Api;
+import io.swagger.annotations.ApiImplicitParam;
import io.swagger.annotations.ApiOperation;
import org.apache.inlong.manager.common.enums.OperationType;
-import org.apache.inlong.manager.pojo.common.Response;
import org.apache.inlong.manager.common.validation.UpdateValidation;
+import org.apache.inlong.manager.pojo.common.PageResult;
+import org.apache.inlong.manager.pojo.common.Response;
import org.apache.inlong.manager.pojo.transform.DeleteTransformRequest;
+import org.apache.inlong.manager.pojo.transform.TransformPageRequest;
import org.apache.inlong.manager.pojo.transform.TransformRequest;
import org.apache.inlong.manager.pojo.transform.TransformResponse;
import org.apache.inlong.manager.service.operationlog.OperationLog;
@@ -30,14 +33,12 @@ import
org.apache.inlong.manager.service.transform.StreamTransformService;
import org.apache.inlong.manager.service.user.LoginUserUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.validation.annotation.Validated;
+import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
-import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
-import java.util.List;
-
/**
* Stream transform control layer
*/
@@ -57,11 +58,17 @@ public class StreamTransformController {
streamTransformService.save(request,
LoginUserUtils.getLoginUser().getName()));
}
- @RequestMapping(value = "/transform/list", method = RequestMethod.GET)
+ @RequestMapping(value = "/transform/list", method = RequestMethod.POST)
@ApiOperation(value = "Get stream transform list")
- public Response<List<TransformResponse>>
list(@RequestParam("inlongGroupId") String groupId,
- @RequestParam("inlongStreamId") String streamId) {
- return Response.success(streamTransformService.listTransform(groupId,
streamId));
+ public Response<PageResult<TransformResponse>> list(@Validated
@RequestBody TransformPageRequest request) {
+ return
Response.success(streamTransformService.listByCondition(request,
LoginUserUtils.getLoginUser()));
+ }
+
+ @RequestMapping(value = "/transform/get/{id}", method = RequestMethod.GET)
+ @ApiOperation(value = "Get stream transform")
+ @ApiImplicitParam(name = "id", dataTypeClass = Integer.class, required =
true)
+ public Response<TransformResponse> get(@PathVariable Integer id) {
+ return Response.success(streamTransformService.get(id,
LoginUserUtils.getLoginUser()));
}
@RequestMapping(value = "/transform/update", method = RequestMethod.POST)
