[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.

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);

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;
+    }
 }

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);

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 {

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

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);
         }
     }
 

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,

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);
     }