This is an automated email from the ASF dual-hosted git repository.
zirui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-inlong.git
The following commit(s) were added to refs/heads/master by this push:
new eb296a1e8 [INLONG-4664][TubeMQ] Add GroupController and
TopicBackendWorker (#4679)
eb296a1e8 is described below
commit eb296a1e8dea1360a4355d12163bd0af19c5bc58
Author: Lizhen <[email protected]>
AuthorDate: Fri Jun 17 17:30:30 2022 +0800
[INLONG-4664][TubeMQ] Add GroupController and TopicBackendWorker (#4679)
---
.../manager/controller/group/GroupController.java | 15 +++
.../group/request/QueryConsumerGroupReq.java} | 28 +++--
.../tubemq/manager/service/NodeServiceImpl.java | 102 ++++++++++++++-
.../tubemq/manager/service/TopicBackendWorker.java | 140 +++++++++++++++++++++
.../inlong/tubemq/manager/service/TopicFuture.java | 59 +++++++++
.../tubemq/manager/service/TopicServiceImpl.java | 13 ++
.../tubemq/manager/service/TubeMQErrorConst.java | 1 +
.../manager/service/interfaces/NodeService.java | 9 ++
.../manager/service/interfaces/TopicService.java | 9 ++
9 files changed, 367 insertions(+), 9 deletions(-)
diff --git
a/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/controller/group/GroupController.java
b/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/controller/group/GroupController.java
index 36d174c24..a5af0fc85 100644
---
a/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/controller/group/GroupController.java
+++
b/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/controller/group/GroupController.java
@@ -29,6 +29,7 @@ import
org.apache.inlong.tubemq.manager.controller.group.request.DeleteBlackGrou
import
org.apache.inlong.tubemq.manager.controller.group.request.DeleteOffsetReq;
import
org.apache.inlong.tubemq.manager.controller.group.request.FilterCondGroupReq;
import
org.apache.inlong.tubemq.manager.controller.group.request.FlowControlGroupReq;
+import
org.apache.inlong.tubemq.manager.controller.group.request.QueryConsumerGroupReq;
import
org.apache.inlong.tubemq.manager.controller.group.request.QueryOffsetReq;
import org.apache.inlong.tubemq.manager.controller.node.request.CloneOffsetReq;
import
org.apache.inlong.tubemq.manager.controller.topic.request.BatchAddGroupAuthReq;
@@ -86,11 +87,25 @@ public class GroupController {
return masterService.baseRequestMaster(gson.fromJson(req,
FilterCondGroupReq.class));
case TubeConst.FLOW_CONTROL:
return masterService.baseRequestMaster(gson.fromJson(req,
FlowControlGroupReq.class));
+ case TubeConst.QUERY:
+ return queryGroupExist(gson.fromJson(req,
QueryConsumerGroupReq.class));
default:
return
TubeMQResult.errorResult(TubeMQErrorConst.NO_SUCH_METHOD);
}
}
+ /**
+ * query group exist
+ * @param req
+ * @return
+ */
+ private TubeMQResult queryGroupExist(QueryConsumerGroupReq req) {
+ if (!req.legal()) {
+ return TubeMQResult.errorResult(TubeMQErrorConst.PARAM_ILLEGAL);
+ }
+ return topicService.queryGroupExist(req);
+ }
+
/**
* add groups in one batch
*
diff --git
a/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/TubeMQErrorConst.java
b/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/controller/group/request/QueryConsumerGroupReq.java
similarity index 56%
copy from
inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/TubeMQErrorConst.java
copy to
inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/controller/group/request/QueryConsumerGroupReq.java
index f59313f00..09a6ece2d 100644
---
a/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/TubeMQErrorConst.java
+++
b/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/controller/group/request/QueryConsumerGroupReq.java
@@ -15,13 +15,25 @@
* limitations under the License.
*/
-package org.apache.inlong.tubemq.manager.service;
+package org.apache.inlong.tubemq.manager.controller.group.request;
-public class TubeMQErrorConst {
- public static final String PARAM_ILLEGAL = "param illegal";
- public static final String BROKER_IN_OTHER_REGION = "resource already
used";
- public static final String RESOURCE_NOT_EXIST = "resource not exsit";
- public static final String MYSQL_ERROR = "mysql error";
- public static final String NO_SUCH_CLUSTER = "no such cluster";
- public static final String NO_SUCH_METHOD = "no such method";
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.ToString;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.inlong.tubemq.manager.controller.node.request.BaseReq;
+
+/**
+ * query consumer group
+ */
+@Data
+@EqualsAndHashCode(callSuper = true)
+@ToString(callSuper = true)
+public class QueryConsumerGroupReq extends BaseReq {
+ private String consumerGroup;
+ private String topicName;
+
+ public boolean legal() {
+ return StringUtils.isNotBlank(topicName) &&
StringUtils.isNotBlank(consumerGroup);
+ }
}
diff --git
a/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/NodeServiceImpl.java
b/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/NodeServiceImpl.java
index 40791ff5d..2e41637df 100644
---
a/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/NodeServiceImpl.java
+++
b/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/NodeServiceImpl.java
@@ -25,7 +25,9 @@ import com.google.gson.Gson;
import java.io.IOException;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
+import java.util.HashSet;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import lombok.extern.slf4j.Slf4j;
@@ -71,9 +73,14 @@ public class NodeServiceImpl implements NodeService {
private final CloseableHttpClient httpclient = HttpClients.createDefault();
private final Gson gson = new Gson();
- @Value("${manager.max.configurable.broker.size:50}")
+ @Value("${manager.max.configurable.broker.size:1}")
private int maxConfigurableBrokerSize;
+ @Value("${manager.max.retry.adding.topic:10}")
+ private int maxRetryAddingTopic;
+
+ private final TopicBackendWorker worker;
+
@Autowired
private MasterRepository masterRepository;
@@ -83,6 +90,10 @@ public class NodeServiceImpl implements NodeService {
@Autowired
private MasterService masterService;
+ public NodeServiceImpl(TopicBackendWorker worker) {
+ this.worker = worker;
+ }
+
/**
* request node status via http.
*
@@ -281,6 +292,95 @@ public class NodeServiceImpl implements NodeService {
} while (end < needReloadList.size());
}
+ /**
+ * handle result, if success, complete it,
+ * if not success, add back to queue without exceeding max retry,
+ * otherwise complete it with exception.
+ *
+ * @param isSuccess
+ * @param topics
+ * @param pendingTopic
+ */
+ private void handleAddingResult(boolean isSuccess, Set<String> topics,
+ Map<String, TopicFuture> pendingTopic) {
+ for (String topic : topics) {
+ TopicFuture future = pendingTopic.get(topic);
+ if (future != null) {
+ if (isSuccess) {
+ future.complete();
+ } else {
+ future.increaseRetryTime();
+ if (future.getRetryTime() > maxRetryAddingTopic) {
+ future.completeExceptional();
+ } else {
+ // add back to queue.
+ worker.addTopicFuture(future);
+ }
+ }
+ }
+ }
+ }
+
+ /**
+ * Adding topic is an async operation, so this method should
+ * 1. check whether pendingTopic contains topic that has failed/succeeded
to be added.
+ * 2. async add topic to tubemq cluster
+ *
+ * @param brokerInfoList - broker list
+ * @param pendingTopic - topicMap
+ */
+ private void handleAddingTopic(MasterEntry masterEntry,
+ TubeHttpBrokerInfoList brokerInfoList,
+ Map<String, TopicFuture> pendingTopic) {
+ // 1. check tubemq cluster by topic name, remove pending topic if has
added.
+ Set<String> brandNewTopics = new HashSet<>();
+ for (String topic : pendingTopic.keySet()) {
+ TubeHttpTopicInfoList topicInfoList =
topicService.requestTopicConfigInfo(masterEntry, topic);
+ if (topicInfoList != null) {
+ // get broker list by topic request
+ List<Integer> topicBrokerList =
topicInfoList.getTopicBrokerIdList();
+ if (topicBrokerList.isEmpty()) {
+ brandNewTopics.add(topic);
+ } else {
+ // remove brokers which have been added.
+ List<Integer> configurableBrokerIdList =
+ brokerInfoList.getConfigurableBrokerIdList();
+ configurableBrokerIdList.removeAll(topicBrokerList);
+ // add topic to satisfy max broker number.
+ Set<String> singleTopic = new HashSet<>();
+ singleTopic.add(topic);
+ int maxBrokers = Math.min(maxConfigurableBrokerSize,
configurableBrokerIdList.size());
+ boolean isSuccess = configBrokersForTopics(masterEntry,
singleTopic,
+ configurableBrokerIdList, maxBrokers);
+ handleAddingResult(isSuccess, singleTopic, pendingTopic);
+ }
+ }
+ }
+ // 2. add new topics to cluster
+ List<Integer> configurableBrokerIdList =
brokerInfoList.getConfigurableBrokerIdList();
+ int maxBrokers = Math.min(maxConfigurableBrokerSize,
configurableBrokerIdList.size());
+ boolean isSuccess = configBrokersForTopics(masterEntry, brandNewTopics,
+ configurableBrokerIdList, maxBrokers);
+ handleAddingResult(isSuccess, brandNewTopics, pendingTopic);
+ }
+
+ @Override
+ public void updateBrokerStatus(int clusterId, Map<String, TopicFuture>
pendingTopic) {
+ MasterEntry masterEntry =
masterRepository.findMasterEntryByClusterIdEquals(clusterId);
+ if (masterEntry != null) {
+ try {
+ TubeHttpBrokerInfoList brokerInfoList =
requestBrokerStatus(masterEntry);
+ if (brokerInfoList != null) {
+ handleAddingTopic(masterEntry, brokerInfoList,
pendingTopic);
+ }
+ } catch (Exception ex) {
+ log.error("exception caught while requesting broker status",
ex);
+ }
+ } else {
+ log.error("cannot get master ip by clusterId {}, please check it",
clusterId);
+ }
+ }
+
@Override
public void close() throws IOException {
httpclient.close();
diff --git
a/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/TopicBackendWorker.java
b/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/TopicBackendWorker.java
new file mode 100644
index 000000000..aaed365bf
--- /dev/null
+++
b/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/TopicBackendWorker.java
@@ -0,0 +1,140 @@
+/*
+ * 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.tubemq.manager.service;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.inlong.tubemq.manager.repository.TopicRepository;
+import org.apache.inlong.tubemq.manager.service.interfaces.NodeService;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Component;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Topic backend thread worker.
+ */
+@Component
+@Slf4j
+public class TopicBackendWorker implements DisposableBean, Runnable {
+ // old code, stop first
+ private final AtomicBoolean runFlag = new AtomicBoolean(false);
+ private final ConcurrentHashMap<Integer, BlockingQueue<TopicFuture>>
pendingTopics =
+ new ConcurrentHashMap<>();
+ private final AtomicInteger notSatisfiedCount = new AtomicInteger(0);
+ private final NodeService nodeService;
+
+ @Autowired
+ private TopicRepository topicRepository;
+
+ @Value("${manager.topic.queue.warning.size:100}")
+ private int queueWarningSize;
+
+ // value in seconds
+ @Value("${manager.topic.queue.thread.interval:10}")
+ private int queueThreadInterval;
+
+ @Value("${manager.topic.queue.max.wait:3}")
+ private int queueMaxWait;
+
+ @Value("${manager.topic.queue.max.running.size:20}")
+ private int queueMaxRunningSize;
+
+ TopicBackendWorker() {
+ Thread thread = new Thread(this);
+ // daemon thread
+ thread.setDaemon(true);
+ thread.start();
+ nodeService = new NodeServiceImpl(this);
+ }
+
+ /**
+ * add topic future to pending executing queue.
+ * @param future - TopicFuture.
+ */
+ public void addTopicFuture(TopicFuture future) {
+ BlockingQueue<TopicFuture> tmpQueue = new LinkedBlockingQueue<>();
+ BlockingQueue<TopicFuture> queue = pendingTopics.putIfAbsent(
+ future.getEntry().getClusterId(), tmpQueue);
+ if (queue == null) {
+ queue = tmpQueue;
+ }
+ queue.add(future);
+ if (queue.size() > queueWarningSize) {
+ log.warn("queue size exceed {}, please check it",
queueWarningSize);
+ }
+ }
+
+ /**
+ * batch executing adding topic, wait util max n seconds or max size
satisfied.
+ */
+ private void batchAddTopic() {
+ pendingTopics.forEach((clusterId, queue) -> {
+ Map<String, TopicFuture> pendingTopicList = new HashMap<>(32);
+ if (notSatisfiedCount.get() > queueMaxWait || queue.size() >
queueMaxRunningSize) {
+ notSatisfiedCount.set(0);
+ List<TopicFuture> tmpTopicList = new ArrayList<>();
+ queue.drainTo(tmpTopicList, queueMaxRunningSize);
+ for (TopicFuture topicFuture : tmpTopicList) {
+ pendingTopicList.put(topicFuture.getEntry().getTopic(),
topicFuture);
+ }
+ } else {
+ notSatisfiedCount.incrementAndGet();
+ }
+ // update broker status
+ nodeService.updateBrokerStatus(clusterId, pendingTopicList);
+ });
+
+ }
+
+ /**
+ * check topic from db
+ */
+ private void checkTopicFromDB() {
+ }
+
+ @Override
+ public void run() {
+ log.info("TopicBackendWorker has started");
+ while (runFlag.get()) {
+ try {
+ batchAddTopic();
+ checkTopicFromDB();
+ TimeUnit.SECONDS.sleep(queueThreadInterval);
+ } catch (Exception exception) {
+ log.warn("exception caught", exception);
+ }
+ }
+ }
+
+ @Override
+ public void destroy() throws Exception {
+ runFlag.set(false);
+ nodeService.close();
+ }
+}
diff --git
a/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/TopicFuture.java
b/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/TopicFuture.java
new file mode 100644
index 000000000..89139fd04
--- /dev/null
+++
b/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/TopicFuture.java
@@ -0,0 +1,59 @@
+/*
+ * 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.tubemq.manager.service;
+
+import lombok.Getter;
+import org.apache.inlong.tubemq.manager.entry.TopicEntry;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * topic business with future.
+ */
+public class TopicFuture {
+ @Getter
+ private int retryTime = 0;
+ @Getter
+ private final TopicEntry entry;
+ @Getter
+ private final CompletableFuture<TopicEntry> future;
+
+ public TopicFuture(TopicEntry entry, CompletableFuture<TopicEntry> future)
{
+ this.entry = entry;
+ this.future = future;
+ }
+
+ /**
+ * record retry time.
+ */
+ public void increaseRetryTime() {
+ retryTime += 1;
+ }
+
+ /**
+ * when topic operation finished, complete it.
+ */
+ public void complete() {
+ this.future.complete(this.entry);
+ }
+
+ public void completeExceptional() {
+ this.future.completeExceptionally(new RuntimeException("exceed max
retry "
+ + retryTime + " adding"));
+ }
+}
diff --git
a/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/TopicServiceImpl.java
b/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/TopicServiceImpl.java
index 49e2dd873..f088815bd 100644
---
a/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/TopicServiceImpl.java
+++
b/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/TopicServiceImpl.java
@@ -33,6 +33,7 @@ import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.inlong.tubemq.manager.controller.TubeMQResult;
import
org.apache.inlong.tubemq.manager.controller.group.request.DeleteOffsetReq;
+import
org.apache.inlong.tubemq.manager.controller.group.request.QueryConsumerGroupReq;
import
org.apache.inlong.tubemq.manager.controller.group.request.QueryOffsetReq;
import
org.apache.inlong.tubemq.manager.controller.group.result.AllBrokersOffsetRes;
import
org.apache.inlong.tubemq.manager.controller.group.result.AllBrokersOffsetRes.OffsetInfo;
@@ -98,6 +99,18 @@ public class TopicServiceImpl implements TopicService {
return null;
}
+ @Override
+ public TubeMQResult queryGroupExist(QueryConsumerGroupReq req) {
+ MasterEntry masterNode = masterService.getMasterNode(req);
+ TubeHttpGroupDetailInfo groupDetailInfo =
requestGroupRunInfo(masterNode,
+ req.getConsumerGroup());
+ List<String> topicSet = groupDetailInfo.getTopicSet();
+ if (topicSet.stream().anyMatch(topic ->
topic.equals(req.getTopicName()))) {
+ return TubeMQResult.successResult();
+ }
+ return TubeMQResult.errorResult(TubeMQErrorConst.NO_SUCH_GROUP);
+ }
+
@Override
public TopicView requestTopicViewInfo(Long clusterId, String topicName) {
MasterEntry masterNode = masterService.getMasterNode(clusterId);
diff --git
a/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/TubeMQErrorConst.java
b/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/TubeMQErrorConst.java
index f59313f00..21b47e3b3 100644
---
a/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/TubeMQErrorConst.java
+++
b/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/TubeMQErrorConst.java
@@ -24,4 +24,5 @@ public class TubeMQErrorConst {
public static final String MYSQL_ERROR = "mysql error";
public static final String NO_SUCH_CLUSTER = "no such cluster";
public static final String NO_SUCH_METHOD = "no such method";
+ public static final String NO_SUCH_GROUP = "no such group";
}
diff --git
a/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/interfaces/NodeService.java
b/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/interfaces/NodeService.java
index a2d76e7a3..f440f3485 100644
---
a/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/interfaces/NodeService.java
+++
b/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/interfaces/NodeService.java
@@ -19,6 +19,7 @@ package org.apache.inlong.tubemq.manager.service.interfaces;
import java.io.IOException;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import org.apache.inlong.tubemq.manager.controller.TubeMQResult;
@@ -28,6 +29,7 @@ import
org.apache.inlong.tubemq.manager.controller.node.request.CloneBrokersReq;
import org.apache.inlong.tubemq.manager.controller.node.request.CloneTopicReq;
import org.apache.inlong.tubemq.manager.entry.ClusterEntry;
import org.apache.inlong.tubemq.manager.entry.MasterEntry;
+import org.apache.inlong.tubemq.manager.service.TopicFuture;
import org.apache.inlong.tubemq.manager.service.tube.TubeHttpBrokerInfoList;
public interface NodeService {
@@ -86,6 +88,13 @@ public interface NodeService {
void handleReloadBroker(MasterEntry masterEntry, List<Integer>
needReloadList, ClusterEntry clusterEntry);
+ /**
+ * update broker status
+ * @param clusterId
+ * @param pendingTopic
+ */
+ void updateBrokerStatus(int clusterId, Map<String, TopicFuture>
pendingTopic);
+
void close() throws IOException;
/**
diff --git
a/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/interfaces/TopicService.java
b/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/interfaces/TopicService.java
index ae38be90a..d40dc45ad 100644
---
a/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/interfaces/TopicService.java
+++
b/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/interfaces/TopicService.java
@@ -19,6 +19,7 @@ package org.apache.inlong.tubemq.manager.service.interfaces;
import org.apache.inlong.tubemq.manager.controller.TubeMQResult;
import
org.apache.inlong.tubemq.manager.controller.group.request.DeleteOffsetReq;
+import
org.apache.inlong.tubemq.manager.controller.group.request.QueryConsumerGroupReq;
import
org.apache.inlong.tubemq.manager.controller.group.request.QueryOffsetReq;
import org.apache.inlong.tubemq.manager.controller.node.request.CloneOffsetReq;
import
org.apache.inlong.tubemq.manager.controller.topic.request.RebalanceGroupReq;
@@ -38,6 +39,14 @@ public interface TopicService {
*/
TubeHttpGroupDetailInfo requestGroupRunInfo(MasterEntry masterEntry,
String group);
+ /**
+ * query if a group exist with a topic
+ *
+ * @param req
+ * @return
+ */
+ TubeMQResult queryGroupExist(QueryConsumerGroupReq req);
+
TopicView requestTopicViewInfo(Long clusterId, String topicName);
/**
