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 f495de5676 [INLONG-9518][Manager] Support resetting the consumption
location of the consumption group used by sort (#9519)
f495de5676 is described below
commit f495de56769b6b99c3e1e7274eea332db113eb23
Author: fuweng11 <[email protected]>
AuthorDate: Wed Dec 27 16:08:04 2023 +0800
[INLONG-9518][Manager] Support resetting the consumption location of the
consumption group used by sort (#9519)
---
.../manager/pojo/consume/SortConsumerInfo.java | 38 ++++++
.../maintenanceTools/MaintenanceToolsService.java | 32 +++++
.../MaintenanceToolsServiceImpl.java | 136 +++++++++++++++++++++
.../resource/queue/QueueResourceOperator.java | 16 +++
.../queue/kafka/KafkaQueueResourceOperator.java | 10 ++
.../resource/queue/pulsar/PulsarOperator.java | 14 +++
.../queue/pulsar/PulsarQueueResourceOperator.java | 40 ++++++
.../service/resource/queue/pulsar/PulsarUtils.java | 9 ++
.../queue/tubemq/TubeMQQueueResourceOperator.java | 9 ++
.../web/controller/MaintenanceToolsController.java | 72 +++++++++++
10 files changed, 376 insertions(+)
diff --git
a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/consume/SortConsumerInfo.java
b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/consume/SortConsumerInfo.java
new file mode 100644
index 0000000000..dd58369ee7
--- /dev/null
+++
b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/consume/SortConsumerInfo.java
@@ -0,0 +1,38 @@
+/*
+ * 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.consume;
+
+import io.swagger.annotations.ApiModel;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+@ApiModel("Sort consumer info")
+public class SortConsumerInfo {
+
+ private Integer sinkId;
+ private String inlongGroupId;
+ private String inlongStreamId;
+ private String consumerGroup;
+
+}
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/maintenanceTools/MaintenanceToolsService.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/maintenanceTools/MaintenanceToolsService.java
new file mode 100644
index 0000000000..1e27771edf
--- /dev/null
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/maintenanceTools/MaintenanceToolsService.java
@@ -0,0 +1,32 @@
+/*
+ * 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.service.maintenanceTools;
+
+import org.apache.inlong.manager.pojo.consume.SortConsumerInfo;
+
+import org.springframework.web.multipart.MultipartFile;
+
+import java.util.List;
+
+public interface MaintenanceToolsService {
+
+ List<SortConsumerInfo> getSortConsumer(MultipartFile file);
+
+ Boolean resetCursor(MultipartFile file, String resetTime);
+
+}
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/maintenanceTools/MaintenanceToolsServiceImpl.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/maintenanceTools/MaintenanceToolsServiceImpl.java
new file mode 100644
index 0000000000..45ec326e5c
--- /dev/null
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/maintenanceTools/MaintenanceToolsServiceImpl.java
@@ -0,0 +1,136 @@
+/*
+ * 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.service.maintenanceTools;
+
+import org.apache.inlong.manager.common.consts.InlongConstants;
+import org.apache.inlong.manager.common.enums.ErrorCodeEnum;
+import org.apache.inlong.manager.common.exceptions.BusinessException;
+import org.apache.inlong.manager.dao.entity.InlongStreamEntity;
+import org.apache.inlong.manager.dao.entity.StreamSinkEntity;
+import org.apache.inlong.manager.dao.mapper.InlongStreamEntityMapper;
+import org.apache.inlong.manager.dao.mapper.StreamSinkEntityMapper;
+import org.apache.inlong.manager.pojo.consume.SortConsumerInfo;
+import org.apache.inlong.manager.pojo.group.InlongGroupInfo;
+import org.apache.inlong.manager.pojo.group.pulsar.InlongPulsarInfo;
+import org.apache.inlong.manager.pojo.user.LoginUserUtils;
+import org.apache.inlong.manager.pojo.user.UserRoleCode;
+import org.apache.inlong.manager.service.group.InlongGroupService;
+import org.apache.inlong.manager.service.resource.queue.QueueResourceOperator;
+import
org.apache.inlong.manager.service.resource.queue.QueueResourceOperatorFactory;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+import org.springframework.web.multipart.MultipartFile;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
+import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+
+@Service
+public class MaintenanceToolsServiceImpl implements MaintenanceToolsService {
+
+ private static final Logger LOGGER =
LoggerFactory.getLogger(MaintenanceToolsServiceImpl.class);
+
+ @Autowired
+ private InlongGroupService groupService;
+ @Autowired
+ private StreamSinkEntityMapper sinkEntityMapper;
+ @Autowired
+ private InlongStreamEntityMapper streamEntityMapper;
+ @Autowired
+ private QueueResourceOperatorFactory queueOperatorFactory;
+
+ @Override
+ public List<SortConsumerInfo> getSortConsumer(MultipartFile file) {
+
LoginUserUtils.getLoginUser().getRoles().add(UserRoleCode.INLONG_SERVICE);
+ List<SortConsumerInfo> sortConsumerInfoList = new ArrayList<>();
+ try (InputStreamReader read = new
InputStreamReader((file.getInputStream()), StandardCharsets.UTF_8)) {
+ BufferedReader bufferedReader = new BufferedReader(read);
+ String readerStr = null;
+ while ((readerStr = bufferedReader.readLine()) != null) {
+ String[] sinkIdList = readerStr.split(InlongConstants.COMMA);
+ for (String sinkIdStr : sinkIdList) {
+ Integer sinkId = Integer.valueOf(sinkIdStr);
+ StreamSinkEntity sinkEntity =
sinkEntityMapper.selectByPrimaryKey(sinkId);
+
+ InlongGroupInfo groupInfo =
groupService.get(sinkEntity.getInlongGroupId());
+ InlongStreamEntity streamEntity = streamEntityMapper
+ .selectByIdentifier(sinkEntity.getInlongGroupId(),
sinkEntity.getInlongStreamId());
+ QueueResourceOperator queueOperator =
queueOperatorFactory.getInstance(groupInfo.getMqType());
+
+ String consumerGroup =
queueOperator.getSortConsumeGroup(groupInfo, streamEntity, sinkEntity);
+
+ SortConsumerInfo sortConsumerInfo =
SortConsumerInfo.builder()
+ .sinkId(sinkId)
+ .consumerGroup(consumerGroup)
+ .inlongGroupId(sinkEntity.getInlongGroupId())
+ .inlongStreamId(sinkEntity.getInlongStreamId())
+ .build();
+ sortConsumerInfoList.add(sortConsumerInfo);
+ }
+ }
+ read.close();
+ LOGGER.info("success get sort consumer");
+ return sortConsumerInfoList;
+ } catch (IOException e) {
+ LOGGER.error("get sort consumer failed:", e);
+ throw new BusinessException(ErrorCodeEnum.INVALID_PARAMETER, "Can
not properly read update file");
+ } finally {
+
LoginUserUtils.getLoginUser().getRoles().remove(UserRoleCode.INLONG_SERVICE);
+ }
+ }
+
+ @Override
+ public Boolean resetCursor(MultipartFile file, String resetTime) {
+
LoginUserUtils.getLoginUser().getRoles().add(UserRoleCode.INLONG_SERVICE);
+ try (InputStreamReader read = new
InputStreamReader((file.getInputStream()), StandardCharsets.UTF_8)) {
+ SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+ Date date = sdf.parse(resetTime);
+ long timeStamp = date.getTime();
+ BufferedReader bufferedReader = new BufferedReader(read);
+ String readerStr = null;
+ while ((readerStr = bufferedReader.readLine()) != null) {
+ String[] sinkIdList = readerStr.split(InlongConstants.COMMA);
+ for (String sinkIdStr : sinkIdList) {
+ Integer sinkId = Integer.valueOf(sinkIdStr);
+ StreamSinkEntity sinkEntity =
sinkEntityMapper.selectByPrimaryKey(sinkId);
+ InlongGroupInfo groupInfo =
groupService.get(sinkEntity.getInlongGroupId());
+ InlongPulsarInfo pulsarInfo = (InlongPulsarInfo) groupInfo;
+ InlongStreamEntity streamEntity = streamEntityMapper
+ .selectByIdentifier(sinkEntity.getInlongGroupId(),
sinkEntity.getInlongStreamId());
+ QueueResourceOperator queueOperator =
queueOperatorFactory.getInstance(groupInfo.getMqType());
+ queueOperator.resetCursor(groupInfo, streamEntity,
sinkEntity, timeStamp);
+ }
+ }
+ read.close();
+ LOGGER.info("success reset cursor consumer");
+ } catch (Exception e) {
+ LOGGER.error("reset cursor consumer failed:", e);
+ throw new BusinessException(ErrorCodeEnum.INVALID_PARAMETER, "Can
not properly read update file");
+ }
+ return true;
+ }
+
+}
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/QueueResourceOperator.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/QueueResourceOperator.java
index 805e92b7fc..4664f670a6 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/QueueResourceOperator.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/QueueResourceOperator.java
@@ -17,6 +17,8 @@
package org.apache.inlong.manager.service.resource.queue;
+import org.apache.inlong.manager.dao.entity.InlongStreamEntity;
+import org.apache.inlong.manager.dao.entity.StreamSinkEntity;
import org.apache.inlong.manager.pojo.consume.BriefMQMessage;
import org.apache.inlong.manager.pojo.group.InlongGroupInfo;
import org.apache.inlong.manager.pojo.stream.InlongStreamInfo;
@@ -88,4 +90,18 @@ public interface QueueResourceOperator {
return null;
}
+ /**
+ * Reset cursor for consumer group
+ *
+ * @param groupInfo inlong group info
+ * @param streamEntity inlong stream entity
+ * @param sinkEntity sink entity
+ * @param resetTime timestamp for reset
+ */
+ default void resetCursor(InlongGroupInfo groupInfo, InlongStreamEntity
streamEntity, StreamSinkEntity sinkEntity,
+ Long resetTime) throws Exception {
+ }
+
+ String getSortConsumeGroup(InlongGroupInfo groupInfo, InlongStreamEntity
streamEntity, StreamSinkEntity sinkEntity);
+
}
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/kafka/KafkaQueueResourceOperator.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/kafka/KafkaQueueResourceOperator.java
index bfc042b6de..cc0fa83dd2 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/kafka/KafkaQueueResourceOperator.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/kafka/KafkaQueueResourceOperator.java
@@ -23,6 +23,8 @@ import org.apache.inlong.manager.common.enums.ClusterType;
import org.apache.inlong.manager.common.enums.ErrorCodeEnum;
import org.apache.inlong.manager.common.exceptions.WorkflowListenerException;
import org.apache.inlong.manager.common.util.Preconditions;
+import org.apache.inlong.manager.dao.entity.InlongStreamEntity;
+import org.apache.inlong.manager.dao.entity.StreamSinkEntity;
import org.apache.inlong.manager.pojo.cluster.ClusterInfo;
import org.apache.inlong.manager.pojo.cluster.kafka.KafkaClusterInfo;
import org.apache.inlong.manager.pojo.consume.BriefMQMessage;
@@ -205,4 +207,12 @@ public class KafkaQueueResourceOperator implements
QueueResourceOperator {
return kafkaOperator.queryLatestMessage((KafkaClusterInfo)
clusterInfo, topicName, consumeGroup, messageCount,
streamInfo);
}
+
+ @Override
+ public String getSortConsumeGroup(InlongGroupInfo groupInfo,
InlongStreamEntity streamEntity,
+ StreamSinkEntity sinkEntity) {
+ InlongKafkaInfo kafkaInfo = (InlongKafkaInfo) groupInfo;
+ String topicName = streamEntity.getMqResource();
+ return String.format(KAFKA_CONSUMER_GROUP,
kafkaInfo.getInlongClusterTag(), topicName);
+ }
}
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarOperator.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarOperator.java
index bdf7b237d8..99653c05c1 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarOperator.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarOperator.java
@@ -467,6 +467,20 @@ public class PulsarOperator {
return briefMQMessages;
}
+ /**
+ * Reset cursor for consumer group.
+ */
+ public void resetCursor(PulsarClusterInfo pulsarClusterInfo, String
topicFullName, String subName,
+ Long resetTime) {
+ try {
+ PulsarUtils.resetCursor(restTemplate, pulsarClusterInfo,
topicFullName, subName,
+ resetTime);
+ } catch (Exception e) {
+ LOGGER.error("failed reset cursor consumer:", e);
+ throw new BusinessException("failed reset cursor consumer:" +
e.getMessage());
+ }
+ }
+
/**
* Build topicName Of Partition
*/
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarQueueResourceOperator.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarQueueResourceOperator.java
index c9d7b4a240..e6e1d66ef5 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarQueueResourceOperator.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarQueueResourceOperator.java
@@ -25,6 +25,8 @@ import org.apache.inlong.manager.common.enums.GroupStatus;
import org.apache.inlong.manager.common.exceptions.BusinessException;
import org.apache.inlong.manager.common.exceptions.WorkflowListenerException;
import org.apache.inlong.manager.common.util.Preconditions;
+import org.apache.inlong.manager.dao.entity.InlongStreamEntity;
+import org.apache.inlong.manager.dao.entity.StreamSinkEntity;
import org.apache.inlong.manager.pojo.cluster.ClusterInfo;
import org.apache.inlong.manager.pojo.cluster.pulsar.PulsarClusterInfo;
import org.apache.inlong.manager.pojo.consume.BriefMQMessage;
@@ -324,4 +326,42 @@ public class PulsarQueueResourceOperator implements
QueueResourceOperator {
id, subs, groupId, topicName);
return briefMQMessages;
}
+
+ /**
+ * Reset cursor for consumer group
+ */
+ public void resetCursor(InlongGroupInfo groupInfo, InlongStreamEntity
streamEntity, StreamSinkEntity sinkEntity,
+ Long resetTime) throws Exception {
+ log.info("begin to reset cursor for sinkId={}", sinkEntity.getId());
+ InlongPulsarInfo pulsarInfo = (InlongPulsarInfo) groupInfo;
+ List<ClusterInfo> clusterInfos =
+
clusterService.listByTagAndType(pulsarInfo.getInlongClusterTag(),
ClusterType.PULSAR);
+ for (ClusterInfo clusterInfo : clusterInfos) {
+ PulsarClusterInfo pulsarCluster = (PulsarClusterInfo) clusterInfo;
+ try {
+ String tenant = pulsarInfo.getPulsarTenant();
+ if (StringUtils.isBlank(tenant)) {
+ tenant = pulsarCluster.getPulsarTenant();
+ }
+ String namespace = pulsarInfo.getMqResource();
+ String topicName = streamEntity.getMqResource();
+ String fullTopicName = tenant + "/" + namespace + "/" +
topicName;
+ String subs = String.format(PULSAR_SUBSCRIPTION,
groupInfo.getInlongClusterTag(), topicName,
+ sinkEntity.getId());
+ pulsarOperator.resetCursor(pulsarCluster, fullTopicName, subs,
resetTime);
+ } catch (Exception e) {
+ log.error("failed reset cursor consumer:", e);
+ throw new BusinessException("failed reset cursor consumer:" +
e.getMessage());
+ }
+ }
+ log.info("success to reset cursor for sinkId={}", sinkEntity.getId());
+ }
+
+ @Override
+ public String getSortConsumeGroup(InlongGroupInfo groupInfo,
InlongStreamEntity streamEntity,
+ StreamSinkEntity sinkEntity) {
+ String topicName = streamEntity.getMqResource();
+ return String.format(PULSAR_SUBSCRIPTION,
groupInfo.getInlongClusterTag(), topicName,
+ sinkEntity.getId());
+ }
}
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarUtils.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarUtils.java
index 64802e7a76..1f60c201bf 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarUtils.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarUtils.java
@@ -787,6 +787,15 @@ public class PulsarUtils {
return ret;
}
+ public static void resetCursor(RestTemplate restTemplate,
PulsarClusterInfo clusterInfo,
+ String topicPath, String subscription, Long resetTime) throws
Exception {
+ HttpUtils.request(restTemplate,
+ clusterInfo.getAdminUrls(QUERY_PERSISTENT_PATH + "/" +
topicPath + "/subscription/"
+ + subscription + "/resetcursor/" + resetTime),
+ HttpMethod.POST, null,
+ getHttpHeaders(clusterInfo.getToken()));
+ }
+
/**
* Copy from deSerializeSingleMessageInBatch method of
org.apache.pulsar.common.protocol.Commands class.
*
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/tubemq/TubeMQQueueResourceOperator.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/tubemq/TubeMQQueueResourceOperator.java
index 849c31828e..f6cf034e67 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/tubemq/TubeMQQueueResourceOperator.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/tubemq/TubeMQQueueResourceOperator.java
@@ -23,6 +23,8 @@ import org.apache.inlong.manager.common.enums.ErrorCodeEnum;
import org.apache.inlong.manager.common.enums.GroupStatus;
import org.apache.inlong.manager.common.exceptions.WorkflowListenerException;
import org.apache.inlong.manager.common.util.Preconditions;
+import org.apache.inlong.manager.dao.entity.InlongStreamEntity;
+import org.apache.inlong.manager.dao.entity.StreamSinkEntity;
import org.apache.inlong.manager.pojo.cluster.tubemq.TubeClusterInfo;
import org.apache.inlong.manager.pojo.consume.BriefMQMessage;
import org.apache.inlong.manager.pojo.group.InlongGroupInfo;
@@ -140,4 +142,11 @@ public class TubeMQQueueResourceOperator implements
QueueResourceOperator {
return tubeMQOperator.queryLastMessage(tubeCluster, topicName,
messageCount, streamInfo);
}
+ @Override
+ public String getSortConsumeGroup(InlongGroupInfo groupInfo,
InlongStreamEntity streamEntity,
+ StreamSinkEntity sinkEntity) {
+ String topicName = streamEntity.getMqResource();
+ return groupInfo.getInlongClusterTag() + "_" + topicName +
"_consumer_group";
+ }
+
}
diff --git
a/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/MaintenanceToolsController.java
b/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/MaintenanceToolsController.java
new file mode 100644
index 0000000000..2565a50704
--- /dev/null
+++
b/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/MaintenanceToolsController.java
@@ -0,0 +1,72 @@
+/*
+ * 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.web.controller;
+
+import org.apache.inlong.manager.pojo.common.Response;
+import org.apache.inlong.manager.pojo.consume.SortConsumerInfo;
+import org.apache.inlong.manager.pojo.user.UserRoleCode;
+import
org.apache.inlong.manager.service.maintenanceTools.MaintenanceToolsService;
+
+import io.swagger.annotations.Api;
+import io.swagger.annotations.ApiImplicitParam;
+import io.swagger.annotations.ApiImplicitParams;
+import io.swagger.annotations.ApiOperation;
+import org.apache.shiro.authz.annotation.RequiresRoles;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.multipart.MultipartFile;
+
+import java.util.List;
+
+/**
+ * Maintenance tools controller.
+ */
+@RestController
+@RequestMapping("/api")
+@Api(tags = "Maintenanct tools-API")
+public class MaintenanceToolsController {
+
+ @Autowired
+ private MaintenanceToolsService maintenanceToolsService;
+
+ @PostMapping("/maintenanceTools/getSortConsumer")
+ @ApiOperation(value = "get sort consumer info")
+ @ApiImplicitParams({
+ @ApiImplicitParam(name = "file", value = "file object", required =
true, dataType = "__FILE", dataTypeClass = MultipartFile.class, paramType =
"query")
+ })
+ @RequiresRoles(value = UserRoleCode.INLONG_ADMIN)
+ public Response<List<SortConsumerInfo>>
getSortConsumer(@RequestParam(value = "file") MultipartFile file) {
+ return Response.success(maintenanceToolsService.getSortConsumer(file));
+ }
+
+ @PostMapping("/maintenanceTools/resetCursor")
+ @ApiOperation(value = "reset cursor consumer")
+ @ApiImplicitParams({
+ @ApiImplicitParam(name = "file", value = "file object", required =
true, dataType = "__FILE", dataTypeClass = MultipartFile.class, paramType =
"query"),
+ @ApiImplicitParam(name = "resetTime", dataTypeClass =
String.class, required = true)
+ })
+ @RequiresRoles(value = UserRoleCode.INLONG_ADMIN)
+ public Response<Boolean> resetCursor(@RequestParam(value = "file")
MultipartFile file,
+ @RequestParam(value = "resetTime") String resetTime) {
+ return Response.success(maintenanceToolsService.resetCursor(file,
resetTime));
+ }
+
+}
