Fix topic deletion for V2 topics. (apache#1470)

* Fix topic deletion for V2 topics.

- Add a isReplicated() method to topics
- Add isReplicated check before deleting a topic

* Remove global check.

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java

@@ -464,7 +464,9 @@ protected void internalUnloadTopic(boolean authoritative) {
     protected void internalDeleteTopic(boolean authoritative) {
         validateAdminOperationOnTopic(authoritative);
         Topic topic = getTopicReference(topicName);
-        if (topicName.isGlobal()) {
+
+        // v2 topics have a global name so check if the topic is replicated.
+        if (topic.isReplicated()) {
             // Delete is disallowed on global topic
             log.error("[{}] Delete topic is forbidden on global namespace {}", clientAppId(), topicName);
             throw new RestException(Status.FORBIDDEN, "Delete forbidden on global namespace");

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java

@@ -111,6 +111,8 @@ CompletableFuture<Consumer> subscribe(ServerCnx cnx, String subscriptionName, lo
 
     boolean isEncryptionRequired();
 
+    boolean isReplicated();
+
     BacklogQuota getBacklogQuota();
 
     void updateRates(NamespaceStats nsStats, NamespaceBundleStats currentBundleStats,

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java

@@ -958,6 +958,11 @@ public boolean isEncryptionRequired() {
         return isEncryptionRequired;
     }
 
+    @Override
+    public boolean isReplicated() {
+        return replicators.size() > 1;
+    }
+
     @Override
     public CompletableFuture<Void> unsubscribe(String subName) {
         // No-op

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

@@ -1497,6 +1497,11 @@ public boolean isEncryptionRequired() {
         return isEncryptionRequired;
     }
 
+    @Override
+    public boolean isReplicated() {
+        return replicators.size() > 1;
+    }
+
     public CompletableFuture<MessageId> terminate() {
         CompletableFuture<MessageId> future = new CompletableFuture<>();
         ledger.asyncTerminate(new TerminateCallback() {