You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by ji...@apache.org on 2022/06/06 06:33:47 UTC
[rocketmq] branch develop updated: [ISSUE#4263] Delete topic route info based on cluster when delete topic. (#4268)
This is an automated email from the ASF dual-hosted git repository.
jinrongtong pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
The following commit(s) were added to refs/heads/develop by this push:
new b1c5fefec [ISSUE#4263] Delete topic route info based on cluster when delete topic. (#4268)
b1c5fefec is described below
commit b1c5fefecf08f802935249e9a05bfd3bb1700fdb
Author: sunxi92 <su...@163.com>
AuthorDate: Mon Jun 6 14:33:35 2022 +0800
[ISSUE#4263] Delete topic route info based on cluster when delete topic. (#4268)
* Delete topic route info based on cluster when delete topic.
* Delete useless compatibility code.
* 1.add a non-null judgment to queueDataMap
2.move the judgment of whether there are any key-value pairs in queueDataMap out of the for loop
* When clusterName is null, then delete topic route info in namesrv.
---
.../rocketmq/client/impl/MQClientAPIImpl.java | 3 ++-
.../DeleteTopicFromNamesrvRequestHeader.java | 10 +++++++
.../namesrv/processor/DefaultRequestProcessor.java | 7 ++++-
.../namesrv/routeinfo/RouteInfoManager.java | 31 ++++++++++++++++++++++
.../rocketmq/tools/admin/DefaultMQAdminExt.java | 7 +++--
.../tools/admin/DefaultMQAdminExtImpl.java | 7 +++--
.../apache/rocketmq/tools/admin/MQAdminExt.java | 3 +--
.../tools/command/topic/DeleteTopicSubCommand.java | 2 +-
8 files changed, 57 insertions(+), 13 deletions(-)
diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java b/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java
index 22aefda8a..57aee0f63 100644
--- a/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java
+++ b/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java
@@ -1494,10 +1494,11 @@ public class MQClientAPIImpl {
throw new MQClientException(response.getCode(), response.getRemark());
}
- public void deleteTopicInNameServer(final String addr, final String topic, final long timeoutMillis)
+ public void deleteTopicInNameServer(final String addr, final String topic, final String clusterName, final long timeoutMillis)
throws RemotingException, InterruptedException, MQClientException {
DeleteTopicFromNamesrvRequestHeader requestHeader = new DeleteTopicFromNamesrvRequestHeader();
requestHeader.setTopic(topic);
+ requestHeader.setClusterName(clusterName);
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.DELETE_TOPIC_IN_NAMESRV, requestHeader);
RemotingCommand response = this.remotingClient.invokeSync(addr, request, timeoutMillis);
diff --git a/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/DeleteTopicFromNamesrvRequestHeader.java b/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/DeleteTopicFromNamesrvRequestHeader.java
index ed8e7a397..c9dd04b60 100644
--- a/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/DeleteTopicFromNamesrvRequestHeader.java
+++ b/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/DeleteTopicFromNamesrvRequestHeader.java
@@ -24,6 +24,8 @@ public class DeleteTopicFromNamesrvRequestHeader implements CommandCustomHeader
@CFNotNull
private String topic;
+ private String clusterName;
+
@Override
public void checkFields() throws RemotingCommandException {
}
@@ -35,4 +37,12 @@ public class DeleteTopicFromNamesrvRequestHeader implements CommandCustomHeader
public void setTopic(String topic) {
this.topic = topic;
}
+
+ public String getClusterName() {
+ return clusterName;
+ }
+
+ public void setClusterName(String clusterName) {
+ this.clusterName = clusterName;
+ }
}
diff --git a/namesrv/src/main/java/org/apache/rocketmq/namesrv/processor/DefaultRequestProcessor.java b/namesrv/src/main/java/org/apache/rocketmq/namesrv/processor/DefaultRequestProcessor.java
index 209e91800..6df8c65bf 100644
--- a/namesrv/src/main/java/org/apache/rocketmq/namesrv/processor/DefaultRequestProcessor.java
+++ b/namesrv/src/main/java/org/apache/rocketmq/namesrv/processor/DefaultRequestProcessor.java
@@ -445,7 +445,12 @@ public class DefaultRequestProcessor extends AsyncNettyRequestProcessor implemen
final DeleteTopicFromNamesrvRequestHeader requestHeader =
(DeleteTopicFromNamesrvRequestHeader) request.decodeCommandCustomHeader(DeleteTopicFromNamesrvRequestHeader.class);
- this.namesrvController.getRouteInfoManager().deleteTopic(requestHeader.getTopic());
+ if (requestHeader.getClusterName() != null
+ && !requestHeader.getClusterName().isEmpty()) {
+ this.namesrvController.getRouteInfoManager().deleteTopic(requestHeader.getTopic(), requestHeader.getClusterName());
+ } else {
+ this.namesrvController.getRouteInfoManager().deleteTopic(requestHeader.getTopic());
+ }
response.setCode(ResponseCode.SUCCESS);
response.setRemark(null);
diff --git a/namesrv/src/main/java/org/apache/rocketmq/namesrv/routeinfo/RouteInfoManager.java b/namesrv/src/main/java/org/apache/rocketmq/namesrv/routeinfo/RouteInfoManager.java
index 49932bc19..400b7e743 100644
--- a/namesrv/src/main/java/org/apache/rocketmq/namesrv/routeinfo/RouteInfoManager.java
+++ b/namesrv/src/main/java/org/apache/rocketmq/namesrv/routeinfo/RouteInfoManager.java
@@ -88,6 +88,37 @@ public class RouteInfoManager {
}
}
+ public void deleteTopic(final String topic, final String clusterName) {
+ try {
+ try {
+ this.lock.writeLock().lockInterruptibly();
+ Set<String> brokerNames = this.clusterAddrTable.get(clusterName);
+ if (brokerNames != null
+ && !brokerNames.isEmpty()) {
+ Map<String, QueueData> queueDataMap = this.topicQueueTable.get(topic);
+ if (queueDataMap != null) {
+ for (String brokerName : brokerNames) {
+ final QueueData removedQD = queueDataMap.remove(brokerName);
+ if (removedQD != null) {
+ log.info("deleteTopic, remove one broker's topic {} {} {}", brokerName, topic,
+ removedQD);
+ }
+ }
+ if (queueDataMap.isEmpty()) {
+ log.info("deleteTopic, remove the topic all queue {} {}", clusterName, topic);
+ this.topicQueueTable.remove(topic);
+ }
+ }
+
+ }
+ } finally {
+ this.lock.writeLock().unlock();
+ }
+ } catch (Exception e) {
+ log.error("deleteTopic Exception", e);
+ }
+ }
+
public TopicList getAllTopicList() {
TopicList topicList = new TopicList();
try {
diff --git a/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExt.java b/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExt.java
index 22b6dcdc7..4088556be 100644
--- a/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExt.java
+++ b/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExt.java
@@ -332,10 +332,9 @@ public class DefaultMQAdminExt extends ClientConfig implements MQAdminExt {
}
@Override
- public void deleteTopicInNameServer(Set<String> addrs,
- String topic) throws RemotingException, MQBrokerException, InterruptedException,
- MQClientException {
- defaultMQAdminExtImpl.deleteTopicInNameServer(addrs, topic);
+ public void deleteTopicInNameServer(final Set<String> addrs, final String topic, final String clusterName) throws RemotingException, MQBrokerException,
+ InterruptedException, MQClientException {
+ defaultMQAdminExtImpl.deleteTopicInNameServer(addrs, topic, clusterName);
}
@Override
diff --git a/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java b/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java
index 292d80cdb..9143c7c5d 100644
--- a/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java
+++ b/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java
@@ -456,15 +456,14 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
}
@Override
- public void deleteTopicInNameServer(Set<String> addrs,
- String topic) throws RemotingException, MQBrokerException, InterruptedException,
- MQClientException {
+ public void deleteTopicInNameServer(Set<String> addrs, String topic, String clusterName) throws RemotingException, MQBrokerException,
+ InterruptedException, MQClientException {
if (addrs == null) {
String ns = this.mqClientInstance.getMQClientAPIImpl().fetchNameServerAddr();
addrs = new HashSet(Arrays.asList(ns.split(";")));
}
for (String addr : addrs) {
- this.mqClientInstance.getMQClientAPIImpl().deleteTopicInNameServer(addr, topic, timeoutMillis);
+ this.mqClientInstance.getMQClientAPIImpl().deleteTopicInNameServer(addr, topic, clusterName, timeoutMillis);
}
}
diff --git a/tools/src/main/java/org/apache/rocketmq/tools/admin/MQAdminExt.java b/tools/src/main/java/org/apache/rocketmq/tools/admin/MQAdminExt.java
index 7bbe8e331..58c6f3f49 100644
--- a/tools/src/main/java/org/apache/rocketmq/tools/admin/MQAdminExt.java
+++ b/tools/src/main/java/org/apache/rocketmq/tools/admin/MQAdminExt.java
@@ -156,8 +156,7 @@ public interface MQAdminExt extends MQAdmin {
void deleteTopicInBroker(final Set<String> addrs, final String topic) throws RemotingException, MQBrokerException,
InterruptedException, MQClientException;
- void deleteTopicInNameServer(final Set<String> addrs,
- final String topic) throws RemotingException, MQBrokerException,
+ void deleteTopicInNameServer(final Set<String> addrs, final String topic, String clusterName) throws RemotingException, MQBrokerException,
InterruptedException, MQClientException;
void deleteSubscriptionGroup(final String addr, String groupName) throws RemotingException, MQBrokerException,
diff --git a/tools/src/main/java/org/apache/rocketmq/tools/command/topic/DeleteTopicSubCommand.java b/tools/src/main/java/org/apache/rocketmq/tools/command/topic/DeleteTopicSubCommand.java
index 6cb3f18b9..aa5a5ead9 100644
--- a/tools/src/main/java/org/apache/rocketmq/tools/command/topic/DeleteTopicSubCommand.java
+++ b/tools/src/main/java/org/apache/rocketmq/tools/command/topic/DeleteTopicSubCommand.java
@@ -48,7 +48,7 @@ public class DeleteTopicSubCommand implements SubCommand {
nameServerSet = new HashSet(Arrays.asList(ns));
}
- adminExt.deleteTopicInNameServer(nameServerSet, topic);
+ adminExt.deleteTopicInNameServer(nameServerSet, topic, clusterName);
System.out.printf("delete topic [%s] from NameServer success.%n", topic);
}