[fix][broker] Don't check replication clusters for ownership of system topics

Author: lhotari

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

@@ -117,6 +117,7 @@
 import org.apache.pulsar.common.policies.data.AuthAction;
 import org.apache.pulsar.common.policies.data.AutoSubscriptionCreationOverride;
 import org.apache.pulsar.common.policies.data.BacklogQuota;
+import org.apache.pulsar.common.policies.data.ClusterDataImpl;
 import org.apache.pulsar.common.policies.data.DelayedDeliveryPolicies;
 import org.apache.pulsar.common.policies.data.DispatchRate;
 import org.apache.pulsar.common.policies.data.EntryFilters;
@@ -4301,19 +4302,27 @@ public static CompletableFuture<PartitionedTopicMetadata> unsafeGetPartitionedTo
         // and other vital information. Even after namespace starting deletion,,
         // we need to access the metadata of system topics to create readers and clean up topic data.
         // If we don't do this, it can prevent namespace deletion due to inaccessible readers.
-        checkLocalOrGetPeerReplicationCluster(pulsar, topicName.getNamespaceObject(), isSystemTopic(topicName))
-            .thenCompose(res -> pulsar.getBrokerService()
-                .fetchPartitionedTopicMetadataCheckAllowAutoCreationAsync(topicName))
-            .thenAccept(metadata -> {
-                if (log.isDebugEnabled()) {
-                    log.debug("Total number of partitions for topic {} is {}", topicName,
-                        metadata.partitions);
-                }
-                metadataFuture.complete(metadata);
-            }).exceptionally(ex -> {
-            metadataFuture.completeExceptionally(ex.getCause());
-            return null;
-        });
+        CompletableFuture<Void> clusterOwnershipCheck;
+        if (isSystemTopic(topicName)) {
+            clusterOwnershipCheck = CompletableFuture.completedFuture(null);
+        } else {
+            clusterOwnershipCheck =
+                    checkLocalOrGetPeerReplicationCluster(pulsar, topicName.getNamespaceObject())
+                            .thenApply(res -> null);
+        }
+
+        clusterOwnershipCheck.thenCompose(res -> pulsar.getBrokerService()
+                        .fetchPartitionedTopicMetadataCheckAllowAutoCreationAsync(topicName))
+                .thenAccept(metadata -> {
+                    if (log.isDebugEnabled()) {
+                        log.debug("Total number of partitions for topic {} is {}", topicName,
+                                metadata.partitions);
+                    }
+                    metadataFuture.complete(metadata);
+                }).exceptionally(ex -> {
+                    metadataFuture.completeExceptionally(ex.getCause());
+                    return null;
+                });
 
         return metadataFuture;
     }