You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by do...@apache.org on 2022/04/25 04:51:37 UTC
[incubator-inlong] branch master updated: [INLONG-3691][TubeMQ] Improve topic adding method and broker replication method and modify broker publish and subscribe parameters (#3897)
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/incubator-inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 874b941ee [INLONG-3691][TubeMQ] Improve topic adding method and broker replication method and modify broker publish and subscribe parameters (#3897)
874b941ee is described below
commit 874b941ee3a32e562f7b5faba6caca3ce5277c50
Author: Lizhen <88...@users.noreply.github.com>
AuthorDate: Mon Apr 25 12:51:32 2022 +0800
[INLONG-3691][TubeMQ] Improve topic adding method and broker replication method and modify broker publish and subscribe parameters (#3897)
Co-authored-by: healchow <he...@gmail.com>
---
.../controller/cluster/ClusterController.java | 2 +-
.../controller/node/request/BatchAddTopicReq.java | 8 ++++++--
.../node/request/BrokerSetReadOrWriteReq.java | 4 ++--
.../controller/topic/TopicWebController.java | 2 +-
.../tubemq/manager/service/NodeServiceImpl.java | 23 +++-------------------
.../manager/service/interfaces/NodeService.java | 9 ---------
.../{IpIdRelation.java => AddTopicRequest.java} | 23 +++++++++++++++++++---
.../tubemq/manager/service/tube/IpIdRelation.java | 10 ++++++++--
8 files changed, 41 insertions(+), 40 deletions(-)
diff --git a/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/controller/cluster/ClusterController.java b/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/controller/cluster/ClusterController.java
index 99c1903e0..939ce33b2 100644
--- a/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/controller/cluster/ClusterController.java
+++ b/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/controller/cluster/ClusterController.java
@@ -252,7 +252,7 @@ public class ClusterController {
String queryMaster = masterService.queryMaster(url);
JsonObject jsonObject = gson.fromJson(queryMaster, JsonObject.class);
JsonElement data = jsonObject.get("data");
- JsonElement dataCount = jsonObject.get("dataCount");
+ JsonElement dataCount = jsonObject.get("count");
Integer topicSize = gson.fromJson(dataCount, Integer.class);
JsonArray jsonData = gson.fromJson(data, JsonArray.class);
int partitionCount = 0;
diff --git a/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/controller/node/request/BatchAddTopicReq.java b/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/controller/node/request/BatchAddTopicReq.java
index fd718e2bc..63480139d 100644
--- a/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/controller/node/request/BatchAddTopicReq.java
+++ b/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/controller/node/request/BatchAddTopicReq.java
@@ -22,11 +22,15 @@ import java.util.List;
import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.ToString;
+import org.apache.inlong.tubemq.manager.service.tube.AddTopicRequest;
@Data
@EqualsAndHashCode(callSuper = true)
@ToString(callSuper = true)
public class BatchAddTopicReq extends BaseReq {
- private List<AddTopicReq> addTopicReqs;
- private List<Integer> brokerIds;
+ private String confModAuthToken;
+
+ private String createUser;
+
+ private List<AddTopicRequest> topicJsonSet;
}
diff --git a/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/controller/node/request/BrokerSetReadOrWriteReq.java b/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/controller/node/request/BrokerSetReadOrWriteReq.java
index 5aeb4ba67..9cc550402 100644
--- a/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/controller/node/request/BrokerSetReadOrWriteReq.java
+++ b/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/controller/node/request/BrokerSetReadOrWriteReq.java
@@ -25,8 +25,8 @@ import lombok.ToString;
@EqualsAndHashCode(callSuper = true)
@ToString(callSuper = true)
public class BrokerSetReadOrWriteReq extends BaseReq {
- private Boolean isAcceptPublish;
- private Boolean isAcceptSubscribe;
+ private Boolean acceptPublish;
+ private Boolean acceptSubscribe;
private String modifyUser;
private String brokerId;
private String confModAuthToken;
diff --git a/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/controller/topic/TopicWebController.java b/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/controller/topic/TopicWebController.java
index 5e9c7967a..e95264902 100644
--- a/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/controller/topic/TopicWebController.java
+++ b/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/controller/topic/TopicWebController.java
@@ -69,7 +69,7 @@ public class TopicWebController {
TubeMQResult topicMethodProxy(@RequestParam String method, @RequestBody String req) throws Exception {
switch (method) {
case TubeConst.ADD:
- return nodeService.batchAddTopic(gson.fromJson(req, BatchAddTopicReq.class));
+ return masterService.baseRequestMaster(gson.fromJson(req, BatchAddTopicReq.class));
case TubeConst.CLONE:
return nodeService.cloneTopicToBrokers(gson.fromJson(req, CloneTopicReq.class));
case TubeConst.AUTH_CONTROL:
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 1edd1a8d5..cefbed2b2 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
@@ -39,7 +39,6 @@ import org.apache.inlong.tubemq.manager.controller.TubeMQResult;
import org.apache.inlong.tubemq.manager.controller.node.dto.MasterDto;
import org.apache.inlong.tubemq.manager.controller.node.request.AddBrokersReq;
import org.apache.inlong.tubemq.manager.controller.node.request.AddTopicReq;
-import org.apache.inlong.tubemq.manager.controller.node.request.BatchAddTopicReq;
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.controller.node.request.QueryBrokerCfgReq;
@@ -125,8 +124,7 @@ public class NodeServiceImpl implements NodeService {
int clusterId = req.getClusterId();
// 1. query source broker config
QueryBrokerCfgReq queryReq = QueryBrokerCfgReq.getReq(req.getSourceBrokerId());
- MasterEntry masterEntry = masterRepository.findMasterEntryByClusterIdEquals(
- clusterId);
+ MasterEntry masterEntry = masterService.getMasterNode(Long.valueOf(clusterId));
BrokerStatusInfo brokerStatusInfo = getBrokerStatusInfo(queryReq, masterEntry);
// 2. use source broker config to clone brokers
@@ -180,7 +178,7 @@ public class NodeServiceImpl implements NodeService {
List<IpIdRelation> ipids = addBrokerResult.getData();
List<Integer> brokerIds = Lists.newArrayList();
for (IpIdRelation ipid : ipids) {
- brokerIds.add(ipid.getId());
+ brokerIds.add(ipid.getBrokerId());
}
return brokerIds;
}
@@ -345,21 +343,6 @@ public class NodeServiceImpl implements NodeService {
}
- /**
- * add topic to brokers
- *
- * @param req
- * @return
- */
- @Override
- public TubeMQResult batchAddTopic(BatchAddTopicReq req) {
- MasterEntry masterEntry = masterService.getMasterNode(req);
- if (masterEntry == null) {
- return TubeMQResult.errorResult(TubeMQErrorConst.NO_SUCH_CLUSTER);
- }
- return addTopicsToBrokers(masterEntry, req.getBrokerIds(), req.getAddTopicReqs());
- }
-
@Override
public void addNode(MasterEntry masterEntry) {
try {
@@ -373,7 +356,7 @@ public class NodeServiceImpl implements NodeService {
@Override
public TubeMQResult modifyMasterNode(MasterDto masterDto) {
try {
- MasterEntry masterEntry = masterRepository.findMasterEntryByClusterIdEquals(masterDto.getClusterId());
+ MasterEntry masterEntry = masterService.getMasterNode(masterDto.getClusterId());
masterEntry.setIp(masterDto.getIp());
masterEntry.setStandby(masterDto.isStandBy());
masterEntry.setToken(masterDto.getToken());
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 e9509c15a..eaa8b176b 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
@@ -24,7 +24,6 @@ import java.util.Set;
import org.apache.inlong.tubemq.manager.controller.TubeMQResult;
import org.apache.inlong.tubemq.manager.controller.node.dto.MasterDto;
import org.apache.inlong.tubemq.manager.controller.node.request.AddTopicReq;
-import org.apache.inlong.tubemq.manager.controller.node.request.BatchAddTopicReq;
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;
@@ -96,14 +95,6 @@ public interface NodeService {
*/
TubeMQResult cloneTopicToBrokers(CloneTopicReq req) throws Exception;
- /**
- * batch add topic to master
- *
- * @param req
- * @return
- */
- TubeMQResult batchAddTopic(BatchAddTopicReq req);
-
/**
* add one node to node repository
*
diff --git a/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/tube/IpIdRelation.java b/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/tube/AddTopicRequest.java
similarity index 58%
copy from inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/tube/IpIdRelation.java
copy to inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/tube/AddTopicRequest.java
index 79295a3d1..bb5c97dca 100644
--- a/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/tube/IpIdRelation.java
+++ b/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/tube/AddTopicRequest.java
@@ -19,8 +19,25 @@ package org.apache.inlong.tubemq.manager.service.tube;
import lombok.Data;
+/**
+ * Request of add topic
+ */
@Data
-public class IpIdRelation {
- private String ip;
- private Integer id;
+public class AddTopicRequest {
+ private Integer brokerId;
+ private String topicName;
+ private String deleteWhen;
+ private String deletePolicy;
+ private Integer numPartitions;
+ private Integer unflushThreshold;
+ private Integer unflushInterval;
+ private Integer unflushDataHold;
+ private Integer numTopicStores;
+ private Integer memCacheMsgCntInK;
+ private Integer memCacheMsgSizeInMB;
+ private Integer memCacheFlushIntvl;
+ private Integer maxMsgSizeInMB;
+ private boolean acceptPublish;
+ private boolean acceptSubscribe;
+ private String createUser;
}
diff --git a/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/tube/IpIdRelation.java b/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/tube/IpIdRelation.java
index 79295a3d1..21a15381c 100644
--- a/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/tube/IpIdRelation.java
+++ b/inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/tube/IpIdRelation.java
@@ -19,8 +19,14 @@ package org.apache.inlong.tubemq.manager.service.tube;
import lombok.Data;
+/**
+ * The relation of broker IP and broker id.
+ */
@Data
public class IpIdRelation {
- private String ip;
- private Integer id;
+ private Integer brokerId;
+ private String brokerIp;
+ private String success;
+ private Integer errCode;
+ private String errInfo;
}