This is an automated email from the ASF dual-hosted git repository.
mikexue 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 d3745519b [ISSUE #3897]Admin, RocketMQAdmin, StandaloneAdmin and
AbstractRmqAdmin can be more clearer. (#3898)
d3745519b is described below
commit d3745519bf5165d5617b8c9bfd9ec5c8addc34fe
Author: pandaapo <[email protected]>
AuthorDate: Tue Jun 6 15:06:39 2023 +0800
[ISSUE #3897]Admin, RocketMQAdmin, StandaloneAdmin and AbstractRmqAdmin can
be more clearer. (#3898)
* Modify Admin's hierarchy.
* Fix CI error.
---
.../apache/eventmesh/api/admin/AbstractAdmin.java} | 44 ++++-------
.../storage/rocketmq/admin/AbstractRmqAdmin.java | 65 ----------------
.../storage/rocketmq/admin/RocketMQAdmin.java | 89 +++++++++++-----------
.../rocketmq/admin/RocketMQAdminAdaptor.java | 4 +-
.../storage/standalone/admin/StandaloneAdmin.java | 33 +-------
5 files changed, 61 insertions(+), 174 deletions(-)
diff --git
a/eventmesh-storage-plugin/eventmesh-storage-rocketmq/src/main/java/org/apache/eventmesh/storage/rocketmq/admin/RocketMQAdminAdaptor.java
b/eventmesh-storage-plugin/eventmesh-storage-api/src/main/java/org/apache/eventmesh/api/admin/AbstractAdmin.java
similarity index 56%
copy from
eventmesh-storage-plugin/eventmesh-storage-rocketmq/src/main/java/org/apache/eventmesh/storage/rocketmq/admin/RocketMQAdminAdaptor.java
copy to
eventmesh-storage-plugin/eventmesh-storage-api/src/main/java/org/apache/eventmesh/api/admin/AbstractAdmin.java
index 27e66d564..5668dbb17 100644
---
a/eventmesh-storage-plugin/eventmesh-storage-rocketmq/src/main/java/org/apache/eventmesh/storage/rocketmq/admin/RocketMQAdminAdaptor.java
+++
b/eventmesh-storage-plugin/eventmesh-storage-api/src/main/java/org/apache/eventmesh/api/admin/AbstractAdmin.java
@@ -15,71 +15,53 @@
* limitations under the License.
*/
-package org.apache.eventmesh.storage.rocketmq.admin;
-
-import org.apache.eventmesh.api.admin.Admin;
-import org.apache.eventmesh.api.admin.TopicProperties;
+package org.apache.eventmesh.api.admin;
import java.util.List;
import java.util.Properties;
+import java.util.concurrent.atomic.AtomicBoolean;
import io.cloudevents.CloudEvent;
-public class RocketMQAdminAdaptor implements Admin {
+public abstract class AbstractAdmin implements Admin {
- private final RocketMQAdmin admin;
+ private final AtomicBoolean started;
- public RocketMQAdminAdaptor() {
- admin = new RocketMQAdmin();
+ public AbstractAdmin(AtomicBoolean started) {
+ this.started = started;
}
@Override
public boolean isStarted() {
- return admin.isStarted();
+ return started.get();
}
@Override
public boolean isClosed() {
- return admin.isClosed();
+ return !started.get();
}
@Override
public void start() {
- admin.start();
+ started.set(true);
}
@Override
public void shutdown() {
- admin.shutdown();
+ started.set(false);
}
@Override
public void init(Properties properties) throws Exception {
- admin.init(properties);
- }
-
- @Override
- public List<TopicProperties> getTopic() throws Exception {
- return admin.getTopic();
- }
- @Override
- public void createTopic(String topicName) throws Exception {
- admin.createTopic(topicName);
- }
-
- @Override
- public void deleteTopic(String topicName) throws Exception {
- admin.deleteTopic(topicName);
}
@Override
- public List<CloudEvent> getEvent(String topicName, int offset, int length)
{
- return admin.getEvent(topicName, offset, length);
+ public List<CloudEvent> getEvent(String topicName, int offset, int length)
throws Exception {
+ return null;
}
@Override
- public void publish(CloudEvent cloudEvent) {
- admin.publish(cloudEvent);
+ public void publish(CloudEvent cloudEvent) throws Exception {
}
}
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
deleted file mode 100644
index 71d5525cc..000000000
---
a/eventmesh-storage-plugin/eventmesh-storage-rocketmq/src/main/java/org/apache/eventmesh/storage/rocketmq/admin/AbstractRmqAdmin.java
+++ /dev/null
@@ -1,65 +0,0 @@
-/*
- * 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.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 3c818be52..af9ff7c37 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
@@ -17,71 +17,66 @@
package org.apache.eventmesh.storage.rocketmq.admin;
-import org.apache.eventmesh.api.admin.Admin;
+import org.apache.eventmesh.api.admin.AbstractAdmin;
import org.apache.eventmesh.api.admin.TopicProperties;
+import org.apache.eventmesh.common.config.ConfigService;
+import org.apache.eventmesh.storage.rocketmq.config.ClientConfiguration;
import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.acl.common.AclClientRPCHook;
+import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.admin.TopicOffset;
import org.apache.rocketmq.common.admin.TopicStatsTable;
import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.remoting.RPCHook;
+import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
import org.apache.rocketmq.tools.command.CommandUtil;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
-import java.util.Properties;
import java.util.Set;
+import java.util.UUID;
import java.util.concurrent.atomic.AtomicBoolean;
-import io.cloudevents.CloudEvent;
-public class RocketMQAdmin extends AbstractRmqAdmin implements Admin {
+public class RocketMQAdmin extends AbstractAdmin {
- private final AtomicBoolean isStarted;
+ private final RPCHook rpcHook;
+
+ protected String nameServerAddr;
+
+ protected String clusterName;
private int numOfQueue = 4;
private int queuePermission = 6;
public RocketMQAdmin() {
- isStarted = new AtomicBoolean(false);
- }
+ super(new AtomicBoolean(false));
- @Override
- public boolean isStarted() {
- return isStarted.get();
- }
-
- @Override
- public boolean isClosed() {
- return !isStarted.get();
- }
-
- @Override
- public void start() {
- isStarted.compareAndSet(false, true);
- }
-
- @Override
- public void shutdown() {
- isStarted.compareAndSet(true, false);
- }
-
- @Override
- public void init(Properties properties) {
+ 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 = new AclClientRPCHook(new SessionCredentials(accessKey,
secretKey));
}
@Override
public List<TopicProperties> getTopic() throws Exception {
+ DefaultMQAdminExt adminExt = createMQAdminExt();
try {
List<TopicProperties> result = new ArrayList<>();
- Set<String> topicList =
getAdminExt().fetchAllTopicList().getTopicList();
+ adminExt.start();
+ Set<String> topicList =
adminExt.fetchAllTopicList().getTopicList();
for (String topic : topicList) {
long messageCount = 0;
- TopicStatsTable topicStats =
getAdminExt().examineTopicStats(topic);
+ TopicStatsTable topicStats = adminExt.examineTopicStats(topic);
HashMap<MessageQueue, TopicOffset> offsetTable =
topicStats.getOffsetTable();
for (TopicOffset topicOffset : offsetTable.values()) {
messageCount += topicOffset.getMaxOffset() -
topicOffset.getMinOffset();
@@ -94,7 +89,7 @@ public class RocketMQAdmin extends AbstractRmqAdmin
implements Admin {
result.sort(Comparator.comparing(t -> t.name));
return result;
} finally {
- shutdownExt();
+ adminExt.shutdown();
}
}
@@ -103,18 +98,20 @@ public class RocketMQAdmin extends AbstractRmqAdmin
implements Admin {
if (StringUtils.isBlank(topicName)) {
throw new Exception("Topic name can not be blank");
}
+ DefaultMQAdminExt adminExt = createMQAdminExt();
try {
- Set<String> brokerAddress =
CommandUtil.fetchMasterAddrByClusterName(getAdminExt(), clusterName);
+ adminExt.start();
+ Set<String> brokerAddress =
CommandUtil.fetchMasterAddrByClusterName(adminExt, clusterName);
for (String masterAddress : brokerAddress) {
TopicConfig topicConfig = new TopicConfig();
topicConfig.setTopicName(topicName);
topicConfig.setReadQueueNums(numOfQueue);
topicConfig.setWriteQueueNums(numOfQueue);
topicConfig.setPerm(queuePermission);
- getAdminExt().createAndUpdateTopicConfig(masterAddress,
topicConfig);
+ adminExt.createAndUpdateTopicConfig(masterAddress,
topicConfig);
}
} finally {
- shutdownExt();
+ adminExt.shutdown();
}
}
@@ -123,22 +120,22 @@ public class RocketMQAdmin extends AbstractRmqAdmin
implements Admin {
if (StringUtils.isBlank(topicName)) {
throw new Exception("Topic name can not be blank.");
}
+ DefaultMQAdminExt adminExt = createMQAdminExt();
try {
- Set<String> brokerAddress =
CommandUtil.fetchMasterAddrByClusterName(getAdminExt(), clusterName);
- getAdminExt().deleteTopicInBroker(brokerAddress, topicName);
+ adminExt.start();
+ Set<String> brokerAddress =
CommandUtil.fetchMasterAddrByClusterName(adminExt, clusterName);
+ adminExt.deleteTopicInBroker(brokerAddress, topicName);
} finally {
- shutdownExt();
+ adminExt.shutdown();
}
}
- @Override
- public List<CloudEvent> getEvent(String topicName, int offset, int length)
{
- return null;
+ private DefaultMQAdminExt createMQAdminExt() {
+ DefaultMQAdminExt adminExt = new DefaultMQAdminExt(rpcHook);
+ String groupId = UUID.randomUUID().toString();
+ adminExt.setAdminExtGroup("admin_ext_group-" + groupId);
+ adminExt.setNamesrvAddr(nameServerAddr);
+ return adminExt;
}
- @Override
- public void publish(CloudEvent cloudEvent) {
- }
-
-
}
diff --git
a/eventmesh-storage-plugin/eventmesh-storage-rocketmq/src/main/java/org/apache/eventmesh/storage/rocketmq/admin/RocketMQAdminAdaptor.java
b/eventmesh-storage-plugin/eventmesh-storage-rocketmq/src/main/java/org/apache/eventmesh/storage/rocketmq/admin/RocketMQAdminAdaptor.java
index 27e66d564..5899cb803 100644
---
a/eventmesh-storage-plugin/eventmesh-storage-rocketmq/src/main/java/org/apache/eventmesh/storage/rocketmq/admin/RocketMQAdminAdaptor.java
+++
b/eventmesh-storage-plugin/eventmesh-storage-rocketmq/src/main/java/org/apache/eventmesh/storage/rocketmq/admin/RocketMQAdminAdaptor.java
@@ -74,12 +74,12 @@ public class RocketMQAdminAdaptor implements Admin {
}
@Override
- public List<CloudEvent> getEvent(String topicName, int offset, int length)
{
+ public List<CloudEvent> getEvent(String topicName, int offset, int length)
throws Exception {
return admin.getEvent(topicName, offset, length);
}
@Override
- public void publish(CloudEvent cloudEvent) {
+ public void publish(CloudEvent cloudEvent) throws Exception {
admin.publish(cloudEvent);
}
}
diff --git
a/eventmesh-storage-plugin/eventmesh-storage-standalone/src/main/java/org/apache/eventmesh/storage/standalone/admin/StandaloneAdmin.java
b/eventmesh-storage-plugin/eventmesh-storage-standalone/src/main/java/org/apache/eventmesh/storage/standalone/admin/StandaloneAdmin.java
index fcaf3a972..ef59f950b 100644
---
a/eventmesh-storage-plugin/eventmesh-storage-standalone/src/main/java/org/apache/eventmesh/storage/standalone/admin/StandaloneAdmin.java
+++
b/eventmesh-storage-plugin/eventmesh-storage-standalone/src/main/java/org/apache/eventmesh/storage/standalone/admin/StandaloneAdmin.java
@@ -17,7 +17,7 @@
package org.apache.eventmesh.storage.standalone.admin;
-import org.apache.eventmesh.api.admin.Admin;
+import org.apache.eventmesh.api.admin.AbstractAdmin;
import org.apache.eventmesh.api.admin.TopicProperties;
import org.apache.eventmesh.storage.standalone.broker.MessageQueue;
import org.apache.eventmesh.storage.standalone.broker.StandaloneBroker;
@@ -26,45 +26,18 @@ import
org.apache.eventmesh.storage.standalone.broker.model.TopicMetadata;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
-import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import io.cloudevents.CloudEvent;
-public class StandaloneAdmin implements Admin {
+public class StandaloneAdmin extends AbstractAdmin {
- private final AtomicBoolean isStarted;
private final StandaloneBroker standaloneBroker;
public StandaloneAdmin() {
+ super(new AtomicBoolean(false));
this.standaloneBroker = StandaloneBroker.getInstance();
- this.isStarted = new AtomicBoolean(false);
- }
-
- @Override
- public boolean isStarted() {
- return isStarted.get();
- }
-
- @Override
- public boolean isClosed() {
- return !isStarted.get();
- }
-
- @Override
- public void start() {
- isStarted.compareAndSet(false, true);
- }
-
- @Override
- public void shutdown() {
- isStarted.compareAndSet(true, false);
- }
-
- @Override
- public void init(Properties properties) throws Exception {
-
}
@Override
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
