[cleanup] PIP-457: Remove NamespaceName.isGlobal() and TopicName.isGlobal()

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java

@@ -597,7 +597,7 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n
                 .thenCompose(__ -> provisionPartitionedTopicPath(numPartitions, createLocalTopicOnly, properties))
                 .thenCompose(__ -> tryCreatePartitionsAsync(numPartitions))
                 .thenRun(() -> {
-                    if (!createLocalTopicOnly && topicName.isGlobal()
+                    if (!createLocalTopicOnly
                             && pulsar().getConfig().isCreateTopicToRemoteClusterForReplication()) {
                         internalCreatePartitionedTopicToReplicatedClustersInBackground(numPartitions);
                         log.info("[{}] Successfully created partitioned for topic {} for the remote clusters",

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/NonPersistentTopics.java

@@ -239,14 +239,12 @@ public void getPartitionedStats(
                 throw new RestException(Response.Status.PRECONDITION_FAILED,
                         "Partitioned Topic Name should not contain '-partition-'");
             }
-            if (topicName.isGlobal()) {
-                try {
-                    validateGlobalNamespaceOwnership(namespaceName);
-                } catch (Exception e) {
-                    log.error("[{}] Failed to get partitioned stats for {}", clientAppId(), topicName, e);
-                    resumeAsyncResponseExceptionally(asyncResponse, e);
-                    return;
-                }
+            try {
+                validateGlobalNamespaceOwnership(namespaceName);
+            } catch (Exception e) {
+                log.error("[{}] Failed to get partitioned stats for {}", clientAppId(), topicName, e);
+                resumeAsyncResponseExceptionally(asyncResponse, e);
+                return;
             }
             getPartitionedTopicMetadataAsync(topicName,
                     authoritative, false).thenAccept(partitionMetadata -> {

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

@@ -2791,9 +2791,6 @@ private void configValueChanged(String configKey, String newValueStr) {
      * @param namespace
      */
     private void unloadDeletedReplNamespace(Policies data, NamespaceName namespace) {
-        if (!namespace.isGlobal()) {
-            return;
-        }
         final String localCluster = this.pulsar.getConfiguration().getClusterName();
         if (pulsar.getBrokerService().isCurrentClusterAllowed(namespace, data)) {
             return;

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

@@ -583,7 +583,7 @@ public CompletableFuture<Void> stopReplProducers() {
     @Override
     public CompletableFuture<Void> checkReplication() {
         TopicName name = TopicName.get(topic);
-        if (!name.isGlobal() || NamespaceService.isHeartbeatNamespace(name)
+        if (NamespaceService.isHeartbeatNamespace(name)
                 || ExtensibleLoadManagerImpl.isInternalTopic(topic)) {
             return CompletableFuture.completedFuture(null);
         }
@@ -990,11 +990,8 @@ public CompletableFuture<PersistentTopicInternalStats> getInternalStats(boolean
     }
 
     public boolean isActive() {
-        if (TopicName.get(topic).isGlobal()) {
-            // No local consumers and no local producers
-            return !subscriptions.isEmpty() || hasLocalProducers();
-        }
-        return currentUsageCount() != 0 || !subscriptions.isEmpty();
+        // No local consumers and no local producers
+        return !subscriptions.isEmpty() || hasLocalProducers();
     }
 
     @Override
@@ -1050,34 +1047,31 @@ public void checkGC() {
         } else {
             if (System.nanoTime() - lastActive > TimeUnit.SECONDS.toNanos(maxInactiveDurationInSec)) {
 
-                if (TopicName.get(topic).isGlobal()) {
-                    // For global namespace, close repl producers first.
-                    // Once all repl producers are closed, we can delete the topic,
-                    // provided no remote producers connected to the broker.
-                    if (log.isDebugEnabled()) {
-                        log.debug("[{}] Global topic inactive for {} seconds, closing repl producers.", topic,
-                            maxInactiveDurationInSec);
-                    }
+                // Close repl producers first.
+                // Once all repl producers are closed, we can delete the topic,
+                // provided no remote producers connected to the broker.
+                if (log.isDebugEnabled()) {
+                    log.debug("[{}] Topic inactive for {} seconds, closing repl producers.", topic,
+                        maxInactiveDurationInSec);
+                }
 
-                    stopReplProducers().thenCompose(v -> delete(true, false))
-                            .thenCompose(__ -> tryToDeletePartitionedMetadata())
-                            .thenRun(() -> log.info("[{}] Topic deleted successfully due to inactivity", topic))
-                            .exceptionally(e -> {
-                                Throwable throwable = e.getCause();
-                                if (throwable instanceof TopicBusyException) {
-                                    // topic became active again
-                                    if (log.isDebugEnabled()) {
-                                        log.debug("[{}] Did not delete busy topic: {}", topic,
-                                                throwable.getMessage());
-                                    }
-                                    replicators.forEach((region, replicator) -> replicator.startProducer());
-                                } else {
-                                    log.warn("[{}] Inactive topic deletion failed", topic, e);
+                stopReplProducers().thenCompose(v -> delete(true, false))
+                        .thenCompose(__ -> tryToDeletePartitionedMetadata())
+                        .thenRun(() -> log.info("[{}] Topic deleted successfully due to inactivity", topic))
+                        .exceptionally(e -> {
+                            Throwable throwable = e.getCause();
+                            if (throwable instanceof TopicBusyException) {
+                                // topic became active again
+                                if (log.isDebugEnabled()) {
+                                    log.debug("[{}] Did not delete busy topic: {}", topic,
+                                            throwable.getMessage());
                                 }
-                                return null;
-                            });
-
-                }
+                                replicators.forEach((region, replicator) -> replicator.startProducer());
+                            } else {
+                                log.warn("[{}] Inactive topic deletion failed", topic, e);
+                            }
+                            return null;
+                        });
             }
         }
     }

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

@@ -1942,7 +1942,7 @@ CompletableFuture<Void> checkPersistencePolicies() {
     @Override
     public CompletableFuture<Void> checkReplication() {
         TopicName name = TopicName.get(topic);
-        if (!name.isGlobal() || NamespaceService.isHeartbeatNamespace(name)
+        if (NamespaceService.isHeartbeatNamespace(name)
                 || ExtensibleLoadManagerImpl.isInternalTopic(topic)) {
             return CompletableFuture.completedFuture(null);
         }
@@ -2112,7 +2112,7 @@ protected CompletableFuture<Boolean> checkAllowedCluster(String localCluster) {
                     if (nsPolicies.isPresent()) {
                         allowedClusters = nsPolicies.get().allowed_clusters;
                     }
-                    if (TopicName.get(topic).isGlobal() && !topicRepls.contains(localCluster)
+                    if (!topicRepls.contains(localCluster)
                             && !allowedClusters.contains(localCluster)) {
                         log.warn("Local cluster {} is not part of global namespace repl list {} and allowed list {}",
                                 localCluster, topicRepls, allowedClusters);
@@ -3197,12 +3197,8 @@ public boolean isActive(InactiveTopicDeleteMode deleteMode) {
                 }
                 break;
         }
-        if (TopicName.get(topic).isGlobal()) {
-            // no local producers
-            return hasLocalProducers();
-        } else {
-            return currentUsageCount() != 0;
-        }
+        // no local producers
+        return hasLocalProducers();
     }
 
     private boolean hasBacklogs(boolean getPreciseBacklog) {
@@ -3410,46 +3406,42 @@ public void checkGC() {
         } else {
             CompletableFuture<Void> replCloseFuture = new CompletableFuture<>();
 
-            if (TopicName.get(topic).isGlobal()) {
-                // For global namespace, close repl producers first.
-                // Once all repl producers are closed, we can delete the topic,
-                // provided no remote producers connected to the broker.
-                if (log.isDebugEnabled()) {
-                    log.debug("[{}] Global topic inactive for {} seconds, closing repl producers.", topic,
-                        maxInactiveDurationInSec);
-                }
-                /**
-                 * There is a race condition that may cause a NPE:
-                 * - task 1: a callback of "replicator.cursor.asyncRead" will trigger a replication.
-                 * - task 2: "closeReplProducersIfNoBacklog" called by current thread will make the variable
-                 *   "replicator.producer" to a null value.
-                 * Race condition: task 1 will get a NPE when it tries to send messages using the variable
-                 * "replicator.producer", because task 2 will set this variable to "null".
-                 * TODO Create a seperated PR to fix it.
-                 */
-                closeReplProducersIfNoBacklog().thenRun(() -> {
-                    if (hasRemoteProducers()) {
-                        if (log.isDebugEnabled()) {
-                            log.debug("[{}] Global topic has connected remote producers. Not a candidate for GC",
-                                    topic);
-                        }
-                        replCloseFuture
-                                .completeExceptionally(new TopicBusyException("Topic has connected remote producers"));
-                    } else {
-                        log.info("[{}] Global topic inactive for {} seconds, closed repl producers", topic,
-                            maxInactiveDurationInSec);
-                        replCloseFuture.complete(null);
-                    }
-                }).exceptionally(e -> {
+            // Close repl producers first.
+            // Once all repl producers are closed, we can delete the topic,
+            // provided no remote producers connected to the broker.
+            if (log.isDebugEnabled()) {
+                log.debug("[{}] Topic inactive for {} seconds, closing repl producers.", topic,
+                    maxInactiveDurationInSec);
+            }
+            /**
+             * There is a race condition that may cause a NPE:
+             * - task 1: a callback of "replicator.cursor.asyncRead" will trigger a replication.
+             * - task 2: "closeReplProducersIfNoBacklog" called by current thread will make the variable
+             *   "replicator.producer" to a null value.
+             * Race condition: task 1 will get a NPE when it tries to send messages using the variable
+             * "replicator.producer", because task 2 will set this variable to "null".
+             * TODO Create a seperated PR to fix it.
+             */
+            closeReplProducersIfNoBacklog().thenRun(() -> {
+                if (hasRemoteProducers()) {
                     if (log.isDebugEnabled()) {
-                        log.debug("[{}] Global topic has replication backlog. Not a candidate for GC", topic);
+                        log.debug("[{}] Topic has connected remote producers. Not a candidate for GC",
+                                topic);
                     }
-                    replCloseFuture.completeExceptionally(e.getCause());
-                    return null;
-                });
-            } else {
-                replCloseFuture.complete(null);
-            }
+                    replCloseFuture
+                            .completeExceptionally(new TopicBusyException("Topic has connected remote producers"));
+                } else {
+                    log.info("[{}] Topic inactive for {} seconds, closed repl producers", topic,
+                        maxInactiveDurationInSec);
+                    replCloseFuture.complete(null);
+                }
+            }).exceptionally(e -> {
+                if (log.isDebugEnabled()) {
+                    log.debug("[{}] Topic has replication backlog. Not a candidate for GC", topic);
+                }
+                replCloseFuture.completeExceptionally(e.getCause());
+                return null;
+            });
 
             replCloseFuture.thenCompose(v -> delete(deleteMode == InactiveTopicDeleteMode.delete_when_no_subscriptions,
                 deleteMode == InactiveTopicDeleteMode.delete_when_subscriptions_caught_up, false))

pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java

@@ -872,7 +872,7 @@ public static CompletableFuture<ClusterDataImpl> checkLocalOrGetPeerReplicationC
     public static CompletableFuture<ClusterDataImpl> checkLocalOrGetPeerReplicationCluster(PulsarService pulsarService,
                                                                                      NamespaceName namespace,
                                                                                      boolean allowDeletedNamespace) {
-        if (!namespace.isGlobal() || NamespaceService.isSLAOrHeartbeatNamespace(namespace.toString())) {
+        if (NamespaceService.isSLAOrHeartbeatNamespace(namespace.toString())) {
             return CompletableFuture.completedFuture(null);
         }
 

pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java

@@ -223,8 +223,7 @@ private void initAndStartBroker() throws Exception {
         admin.clusters().createCluster("use", ClusterData.builder().serviceUrl(pulsar.getWebServiceAddress()).build());
         admin.clusters().createCluster("usw", ClusterData.builder().serviceUrl("http://127.0.0.2:8082").build());
         admin.clusters().createCluster("usc", ClusterData.builder().serviceUrl("http://127.0.0.3:8083").build());
-        // After V1 removal, all namespaces go through the peer-cluster redirect path
-        // (NamespaceName.isGlobal() always returns true), so peer clusters must be configured.
+        // All namespaces go through the peer-cluster redirect path, so peer clusters must be configured.
         // Only "usc" is a peer because peer clusters cannot also be replication clusters.
         admin.clusters().updatePeerClusterNames("use",
                 new LinkedHashSet<>(List.of("usc")));

pulsar-common/src/main/java/org/apache/pulsar/common/naming/NamespaceName.java

@@ -119,10 +119,6 @@ public String getLocalName() {
         return localName;
     }
 
-    public boolean isGlobal() {
-        return true;
-    }
-
     public String getPersistentTopicName(String localTopic) {
         return getTopicName(TopicDomain.persistent, localTopic);
     }

pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java

@@ -352,10 +352,6 @@ public String getLookupName() {
         return String.format("%s/%s/%s/%s", domain, tenant, namespacePortion, getEncodedLocalName());
     }
 
-    public boolean isGlobal() {
-        return namespaceName.isGlobal();
-    }
-
     public String getSchemaName() {
         return getTenant()
             + "/" + getNamespacePortion()

pulsar-common/src/test/java/org/apache/pulsar/common/naming/NamespaceNameTest.java

@@ -20,7 +20,6 @@
 
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertNotEquals;
-import static org.testng.Assert.assertTrue;
 import org.testng.annotations.Test;
 
 public class NamespaceNameTest {
@@ -114,7 +113,6 @@ void testNamespaceProperties() {
         NamespaceName ns = NamespaceName.get("my-tenant/my-namespace");
         assertEquals(ns.getTenant(), "my-tenant");
         assertEquals(ns.getLocalName(), "my-namespace");
-        assertTrue(ns.isGlobal());
         assertEquals(ns.getPersistentTopicName("my-topic"), "persistent://my-tenant/my-namespace/my-topic");
     }
 }