This is an automated email from the ASF dual-hosted git repository.
chenguangsheng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/eventmesh.git
The following commit(s) were added to refs/heads/master by this push:
new 81c4e07dd fix issues #3894
new b6d9befb5 Merge pull request #3896 from g0715158/master
81c4e07dd is described below
commit 81c4e07dd5e3a5b72c08b5eb2755090e10c13337
Author: g0715158 <[email protected]>
AuthorDate: Wed May 10 21:16:26 2023 +0800
fix issues #3894
---
.../storage/rocketmq/admin/AbstractRmqAdmin.java | 47 ++++++++++++++++++++++
.../storage/rocketmq/admin/RocketMQAdmin.java | 47 ++++++----------------
2 files changed, 60 insertions(+), 34 deletions(-)
diff --git
a/eventmesh-storage-plugin/eventmesh-storage-rocketmq/src/main/java/org/apache/eventmesh/storage/rocketmq/admin/AbstractRmqAdmin.java
b/eventmesh-storage-plugin/eventmesh-storage-rocketmq/src/main/java/org/apache/eventmesh/storage/rocketmq/admin/AbstractRmqAdmin.java
new file mode 100644
index 000000000..b8cfd0076
--- /dev/null
+++
b/eventmesh-storage-plugin/eventmesh-storage-rocketmq/src/main/java/org/apache/eventmesh/storage/rocketmq/admin/AbstractRmqAdmin.java
@@ -0,0 +1,47 @@
+package org.apache.eventmesh.storage.rocketmq.admin;
+
+import org.apache.eventmesh.common.config.ConfigService;
+import org.apache.eventmesh.storage.rocketmq.config.ClientConfiguration;
+import org.apache.rocketmq.acl.common.AclClientRPCHook;
+import org.apache.rocketmq.acl.common.SessionCredentials;
+import org.apache.rocketmq.remoting.RPCHook;
+import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
+
+import java.util.UUID;
+
+public abstract class AbstractRmqAdmin {
+ private DefaultMQAdminExt adminExt;
+
+
+ protected String nameServerAddr;
+
+ protected String clusterName;
+
+ protected DefaultMQAdminExt getAdminExt() throws Exception {
+ ConfigService configService = ConfigService.getInstance();
+
+ ClientConfiguration clientConfiguration =
configService.buildConfigInstance(ClientConfiguration.class);
+
+ nameServerAddr = clientConfiguration.getNamesrvAddr();
+ clusterName = clientConfiguration.getClusterName();
+ String accessKey = clientConfiguration.getAccessKey();
+ String secretKey = clientConfiguration.getSecretKey();
+ if (adminExt == null) {
+
+ RPCHook rpcHook = new AclClientRPCHook(new
SessionCredentials(accessKey, secretKey));
+ adminExt = new DefaultMQAdminExt(rpcHook);
+ String groupId = UUID.randomUUID().toString();
+ adminExt.setAdminExtGroup("admin_ext_group-" + groupId);
+ adminExt.setNamesrvAddr(nameServerAddr);
+ adminExt.start();
+ }
+
+ return adminExt;
+ }
+
+ protected void shutdownExt() {
+ adminExt.shutdown();
+ adminExt = null;
+ }
+
+}
diff --git
a/eventmesh-storage-plugin/eventmesh-storage-rocketmq/src/main/java/org/apache/eventmesh/storage/rocketmq/admin/RocketMQAdmin.java
b/eventmesh-storage-plugin/eventmesh-storage-rocketmq/src/main/java/org/apache/eventmesh/storage/rocketmq/admin/RocketMQAdmin.java
index ef30608ce..0aaae3e04 100644
---
a/eventmesh-storage-plugin/eventmesh-storage-rocketmq/src/main/java/org/apache/eventmesh/storage/rocketmq/admin/RocketMQAdmin.java
+++
b/eventmesh-storage-plugin/eventmesh-storage-rocketmq/src/main/java/org/apache/eventmesh/storage/rocketmq/admin/RocketMQAdmin.java
@@ -46,35 +46,15 @@ import java.util.concurrent.atomic.AtomicBoolean;
import io.cloudevents.CloudEvent;
-public class RocketMQAdmin implements Admin {
+public class RocketMQAdmin extends AbstractRmqAdmin implements Admin {
private final AtomicBoolean isStarted;
- protected DefaultMQAdminExt adminExt;
-
- protected String nameServerAddr;
-
- protected String clusterName;
-
private int numOfQueue = 4;
private int queuePermission = 6;
public RocketMQAdmin() {
isStarted = new AtomicBoolean(false);
-
- ConfigService configService = ConfigService.getInstance();
- ClientConfiguration clientConfiguration =
configService.buildConfigInstance(ClientConfiguration.class);
-
- nameServerAddr = clientConfiguration.getNamesrvAddr();
- clusterName = clientConfiguration.getClusterName();
- String accessKey = clientConfiguration.getAccessKey();
- String secretKey = clientConfiguration.getSecretKey();
-
- RPCHook rpcHook = new AclClientRPCHook(new
SessionCredentials(accessKey, secretKey));
- adminExt = new DefaultMQAdminExt(rpcHook);
- String groupId = UUID.randomUUID().toString();
- adminExt.setAdminExtGroup("admin_ext_group-" + groupId);
- adminExt.setNamesrvAddr(nameServerAddr);
}
@Override
@@ -105,26 +85,25 @@ public class RocketMQAdmin implements Admin {
@Override
public List<TopicProperties> getTopic() throws Exception {
try {
- adminExt.start();
List<TopicProperties> result = new ArrayList<>();
- Set<String> topicList =
adminExt.fetchAllTopicList().getTopicList();
+ Set<String> topicList =
getAdminExt().fetchAllTopicList().getTopicList();
for (String topic : topicList) {
long messageCount = 0;
- TopicStatsTable topicStats = adminExt.examineTopicStats(topic);
+ TopicStatsTable topicStats =
getAdminExt().examineTopicStats(topic);
HashMap<MessageQueue, TopicOffset> offsetTable =
topicStats.getOffsetTable();
for (TopicOffset topicOffset : offsetTable.values()) {
messageCount += topicOffset.getMaxOffset() -
topicOffset.getMinOffset();
}
result.add(new TopicProperties(
- topic, messageCount
+ topic, messageCount
));
}
result.sort(Comparator.comparing(t -> t.name));
return result;
} finally {
- adminExt.shutdown();
+ shutdownExt();
}
}
@@ -134,18 +113,17 @@ public class RocketMQAdmin implements Admin {
throw new Exception("Topic name can not be blank");
}
try {
- adminExt.start();
- Set<String> brokerAddress =
CommandUtil.fetchMasterAddrByClusterName(adminExt, clusterName);
+ Set<String> brokerAddress =
CommandUtil.fetchMasterAddrByClusterName(getAdminExt(), clusterName);
for (String masterAddress : brokerAddress) {
TopicConfig topicConfig = new TopicConfig();
topicConfig.setTopicName(topicName);
topicConfig.setReadQueueNums(numOfQueue);
topicConfig.setWriteQueueNums(numOfQueue);
topicConfig.setPerm(queuePermission);
- adminExt.createAndUpdateTopicConfig(masterAddress,
topicConfig);
+ getAdminExt().createAndUpdateTopicConfig(masterAddress,
topicConfig);
}
} finally {
- adminExt.shutdown();
+ shutdownExt();
}
}
@@ -155,11 +133,10 @@ public class RocketMQAdmin implements Admin {
throw new Exception("Topic name can not be blank.");
}
try {
- adminExt.start();
- Set<String> brokerAddress =
CommandUtil.fetchMasterAddrByClusterName(adminExt, clusterName);
- adminExt.deleteTopicInBroker(brokerAddress, topicName);
+ Set<String> brokerAddress =
CommandUtil.fetchMasterAddrByClusterName(getAdminExt(), clusterName);
+ getAdminExt().deleteTopicInBroker(brokerAddress, topicName);
} finally {
- adminExt.shutdown();
+ shutdownExt();
}
}
@@ -171,4 +148,6 @@ public class RocketMQAdmin implements Admin {
@Override
public void publish(CloudEvent cloudEvent) {
}
+
+
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
