From 4b986d3ddcbf152d61b527e602e34973249fcf72 Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Fri, 13 Mar 2026 15:12:21 -0700 Subject: [PATCH] [cleanup] PIP-457: Remove NamespaceName.isGlobal() and TopicName.isGlobal() With V1 namespace removal, isGlobal() always returns true, making it dead code. Removed the methods and simplified all call sites: - Always-true conditions unwrapped to just the body - Always-false conditions (negated) removed as dead code - Combined conditions simplified by removing the isGlobal() term --- .../pulsar/broker/admin/AdminResource.java | 2 +- .../admin/impl/PersistentTopicsBase.java | 349 +++++------------- .../broker/admin/v2/NonPersistentTopics.java | 14 +- .../pulsar/broker/service/BrokerService.java | 3 - .../nonpersistent/NonPersistentTopic.java | 58 ++- .../service/persistent/PersistentTopic.java | 84 ++--- .../pulsar/broker/web/PulsarWebResource.java | 2 +- .../pulsar/broker/admin/NamespacesTest.java | 3 +- .../pulsar/common/naming/NamespaceName.java | 4 - .../pulsar/common/naming/TopicName.java | 4 - .../common/naming/NamespaceNameTest.java | 2 - 11 files changed, 169 insertions(+), 356 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java index ea8e584c6e398..190ed01114329 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java +++ b/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", diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java index fed59cf06e217..ad6854e24260b 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java @@ -328,11 +328,9 @@ protected void internalRevokePermissionsOnTopic(AsyncResponse asyncResponse, Str protected CompletableFuture internalCreateNonPartitionedTopicAsync(boolean authoritative, Map properties) { - CompletableFuture ret = validateNonPartitionTopicNameAsync(topicName.getLocalName()); - if (topicName.isGlobal()) { - ret = ret.thenCompose(__ -> validateGlobalNamespaceOwnershipAsync(namespaceName)); - } - return ret.thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative)) + return validateNonPartitionTopicNameAsync(topicName.getLocalName()) + .thenCompose(__ -> validateGlobalNamespaceOwnershipAsync(namespaceName)) + .thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative)) .thenCompose(__ -> validateNamespaceOperationAsync(topicName.getNamespaceObject(), NamespaceOperation.CREATE_TOPIC)) .thenCompose(__ -> getPartitionedTopicMetadataAsync(topicName, false, false)) @@ -472,7 +470,7 @@ protected CompletableFuture internalCreateNonPartitionedTopicAsync(boolean }) ); }).thenCompose(__ -> { - if (updateLocal || !topicName.isGlobal()) { + if (updateLocal) { return CompletableFuture.completedFuture(null); } // update remote cluster @@ -531,9 +529,9 @@ private CompletableFuture> getReplicationClusters() { protected void internalCreateMissedPartitions(AsyncResponse asyncResponse) { getPartitionedTopicMetadataAsync(topicName, false, false).thenAccept(metadata -> { if (metadata != null && metadata.partitions > 0) { - CompletableFuture future = validateNamespaceOperationAsync(topicName.getNamespaceObject(), - NamespaceOperation.CREATE_TOPIC); - future.thenCompose(__ -> tryCreatePartitionsAsync(metadata.partitions)).thenAccept(v -> { + validateNamespaceOperationAsync(topicName.getNamespaceObject(), + NamespaceOperation.CREATE_TOPIC) + .thenCompose(__ -> tryCreatePartitionsAsync(metadata.partitions)).thenAccept(v -> { asyncResponse.resume(Response.noContent().build()); }).exceptionally(e -> { log.error("[{}] Failed to create partitions for topic {}", clientAppId(), topicName); @@ -907,13 +905,9 @@ private CompletableFuture internalRemovePartitionsAuthenticationPoliciesAs protected void internalUnloadTopic(AsyncResponse asyncResponse, boolean authoritative) { log.info("[{}] Unloading topic {}", clientAppId(), topicName); - CompletableFuture future = validateTopicOperationAsync(topicName, TopicOperation.UNLOAD); - future.thenCompose(__ -> { - if (topicName.isGlobal()) { - return validateGlobalNamespaceOwnershipAsync(namespaceName); - } - return CompletableFuture.completedFuture(null); - }).thenAccept(__ -> { + validateTopicOperationAsync(topicName, TopicOperation.UNLOAD) + .thenCompose(__ -> validateGlobalNamespaceOwnershipAsync(namespaceName)) + .thenAccept(__ -> { // If the topic name is a partition name, no need to get partition topic metadata again if (topicName.isPartitioned()) { if (isTransactionCoordinatorAssign(topicName)) { @@ -1182,14 +1176,9 @@ private boolean isUnexpectedTopicName(PartitionedTopicMetadata topicMetadata) { } protected void internalGetSubscriptions(AsyncResponse asyncResponse, boolean authoritative) { - CompletableFuture future = validateTopicOperationAsync(topicName, TopicOperation.GET_SUBSCRIPTIONS); - future.thenCompose(__ -> { - if (topicName.isGlobal()) { - return validateGlobalNamespaceOwnershipAsync(namespaceName); - } else { - return CompletableFuture.completedFuture(null); - } - }).thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative)) + validateTopicOperationAsync(topicName, TopicOperation.GET_SUBSCRIPTIONS) + .thenCompose(__ -> validateGlobalNamespaceOwnershipAsync(namespaceName)) + .thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative)) .thenAccept(__ -> { // If the topic name is a partition name, no need to get partition topic metadata again if (topicName.isPartitioned()) { @@ -1312,28 +1301,18 @@ private void internalGetSubscriptionsForNonPartitionedTopic(AsyncResponse asyncR protected CompletableFuture internalGetStatsAsync(boolean authoritative, GetStatsOptions getStatsOptions) { - CompletableFuture future = validateTopicOperationAsync(topicName, TopicOperation.GET_STATS); - return future.thenCompose(__ -> { - if (topicName.isGlobal()) { - return validateGlobalNamespaceOwnershipAsync(namespaceName); - } else { - return CompletableFuture.completedFuture(null); - } - }).thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative)) + return validateTopicOperationAsync(topicName, TopicOperation.GET_STATS) + .thenCompose(__ -> validateGlobalNamespaceOwnershipAsync(namespaceName)) + .thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative)) .thenCompose(__ -> getTopicReferenceAsync(topicName)) .thenCompose(topic -> topic.asyncGetStats(getStatsOptions)); } protected CompletableFuture internalGetInternalStatsAsync(boolean authoritative, boolean metadata) { - CompletableFuture future = validateTopicOperationAsync(topicName, TopicOperation.GET_STATS); - return future.thenCompose(__ -> { - if (topicName.isGlobal()) { - return validateGlobalNamespaceOwnershipAsync(namespaceName); - } else { - return CompletableFuture.completedFuture(null); - } - }).thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative)) + return validateTopicOperationAsync(topicName, TopicOperation.GET_STATS) + .thenCompose(__ -> validateGlobalNamespaceOwnershipAsync(namespaceName)) + .thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative)) .thenCompose(__ -> { if (metadata) { return validateTopicOperationAsync(topicName, TopicOperation.GET_METADATA); @@ -1345,13 +1324,9 @@ protected CompletableFuture internalGetInternalSta } protected void internalGetManagedLedgerInfo(AsyncResponse asyncResponse, boolean authoritative) { - CompletableFuture future = validateTopicOperationAsync(topicName, TopicOperation.GET_STATS); - future.thenCompose(__ -> { - if (topicName.isGlobal()) { - return validateGlobalNamespaceOwnershipAsync(namespaceName); - } - return CompletableFuture.completedFuture(null); - }).thenAccept(__ -> { + validateTopicOperationAsync(topicName, TopicOperation.GET_STATS) + .thenCompose(__ -> validateGlobalNamespaceOwnershipAsync(namespaceName)) + .thenAccept(__ -> { // If the topic name is a partition name, no need to get partition topic metadata again if (topicName.isPartitioned()) { internalGetManagedLedgerInfoForNonPartitionedTopic(asyncResponse); @@ -1463,13 +1438,9 @@ public void getInfoFailed(ManagedLedgerException exception, Object ctx) { protected void internalGetPartitionedStats(AsyncResponse asyncResponse, boolean authoritative, boolean perPartition, GetStatsOptions getStatsOptions) { - CompletableFuture future = validateTopicOperationAsync(topicName, TopicOperation.GET_STATS); - future.thenCompose(__ -> { - if (topicName.isGlobal()) { - return validateGlobalNamespaceOwnershipAsync(namespaceName); - } - return CompletableFuture.completedFuture(null); - }).thenCompose(__ -> getPartitionedTopicMetadataAsync(topicName, + validateTopicOperationAsync(topicName, TopicOperation.GET_STATS) + .thenCompose(__ -> validateGlobalNamespaceOwnershipAsync(namespaceName)) + .thenCompose(__ -> getPartitionedTopicMetadataAsync(topicName, authoritative, false)).thenAccept(partitionMetadata -> { if (partitionMetadata.partitions == 0) { asyncResponse.resume(new RestException(Status.NOT_FOUND, @@ -1553,14 +1524,9 @@ protected void internalGetPartitionedStats(AsyncResponse asyncResponse, boolean } protected void internalGetPartitionedStatsInternal(AsyncResponse asyncResponse, boolean authoritative) { - CompletableFuture future = validateTopicOperationAsync(topicName, TopicOperation.GET_STATS); - future.thenCompose(__ -> { - if (topicName.isGlobal()) { - return validateGlobalNamespaceOwnershipAsync(namespaceName); - } else { - return CompletableFuture.completedFuture(null); - } - }).thenCompose(__ -> getPartitionedTopicMetadataAsync(topicName, authoritative, false)) + validateTopicOperationAsync(topicName, TopicOperation.GET_STATS) + .thenCompose(__ -> validateGlobalNamespaceOwnershipAsync(namespaceName)) + .thenCompose(__ -> getPartitionedTopicMetadataAsync(topicName, authoritative, false)) .thenAccept(partitionMetadata -> { if (partitionMetadata.partitions == 0) { asyncResponse.resume(new RestException(Status.NOT_FOUND, @@ -1611,14 +1577,9 @@ protected void internalGetPartitionedStatsInternal(AsyncResponse asyncResponse, protected CompletableFuture internalDeleteSubscriptionAsync(String subName, boolean authoritative, boolean force) { - CompletableFuture future = validateTopicOperationAsync(topicName, TopicOperation.UNSUBSCRIBE, subName); - return future.thenCompose(__ -> { - if (topicName.isGlobal()) { - return validateGlobalNamespaceOwnershipAsync(namespaceName); - } else { - return CompletableFuture.completedFuture(null); - } - }).thenCompose(__ -> { + return validateTopicOperationAsync(topicName, TopicOperation.UNSUBSCRIBE, subName) + .thenCompose(__ -> validateGlobalNamespaceOwnershipAsync(namespaceName)) + .thenCompose(__ -> { if (topicName.isPartitioned()) { return internalDeleteSubscriptionForNonPartitionedTopicAsync(subName, authoritative, force); } else { @@ -1790,14 +1751,9 @@ private void internalGetSubscriptionPropertiesForNonPartitionedTopic(AsyncRespon } protected void internalSkipAllMessages(AsyncResponse asyncResponse, String subName, boolean authoritative) { - CompletableFuture future = validateTopicOperationAsync(topicName, TopicOperation.SKIP, subName); - future.thenCompose(__ -> { - if (topicName.isGlobal()) { - return validateGlobalNamespaceOwnershipAsync(namespaceName); - } else { - return CompletableFuture.completedFuture(null); - } - }).thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative)) + validateTopicOperationAsync(topicName, TopicOperation.SKIP, subName) + .thenCompose(__ -> validateGlobalNamespaceOwnershipAsync(namespaceName)) + .thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative)) .thenCompose(__ -> { // If the topic name is a partition name, no need to get partition topic metadata again if (topicName.isPartitioned()) { @@ -1903,14 +1859,9 @@ private CompletableFuture internalSkipAllMessagesForNonPartitionedTopicAsy protected void internalSkipMessages(AsyncResponse asyncResponse, String subName, int numMessages, boolean authoritative) { - CompletableFuture future = validateTopicOperationAsync(topicName, TopicOperation.SKIP, subName); - future.thenCompose(__ -> { - if (topicName.isGlobal()) { - return validateGlobalNamespaceOwnershipAsync(namespaceName); - } else { - return CompletableFuture.completedFuture(null); - } - }).thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative)) + validateTopicOperationAsync(topicName, TopicOperation.SKIP, subName) + .thenCompose(__ -> validateGlobalNamespaceOwnershipAsync(namespaceName)) + .thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative)) .thenCompose(__ -> getPartitionedTopicMetadataAsync(topicName, authoritative, false)) .thenCompose(partitionMetadata -> { if (partitionMetadata.partitions > 0) { @@ -1967,14 +1918,9 @@ protected void internalSkipMessages(AsyncResponse asyncResponse, String subName, protected void internalExpireMessagesForAllSubscriptions(AsyncResponse asyncResponse, int expireTimeInSeconds, boolean authoritative) { - CompletableFuture future = validateTopicOperationAsync(topicName, TopicOperation.EXPIRE_MESSAGES); - future.thenCompose(__ -> { - if (topicName.isGlobal()) { - return validateGlobalNamespaceOwnershipAsync(namespaceName); - } else { - return CompletableFuture.completedFuture(null); - } - }).thenCompose(__ -> getPartitionedTopicMetadataAsync(topicName, authoritative, false)) + validateTopicOperationAsync(topicName, TopicOperation.EXPIRE_MESSAGES) + .thenCompose(__ -> validateGlobalNamespaceOwnershipAsync(namespaceName)) + .thenCompose(__ -> getPartitionedTopicMetadataAsync(topicName, authoritative, false)) .thenAccept(partitionMetadata -> { if (topicName.isPartitioned()) { internalExpireMessagesForAllSubscriptionsForNonPartitionedTopic(asyncResponse, @@ -2102,14 +2048,9 @@ private void internalExpireMessagesForAllSubscriptionsForNonPartitionedTopic(Asy protected CompletableFuture internalResetCursorAsync(String subName, long timestamp, boolean authoritative) { - CompletableFuture future = validateTopicOperationAsync(topicName, TopicOperation.RESET_CURSOR, subName); - return future.thenCompose(__ -> { - if (topicName.isGlobal()) { - return validateGlobalNamespaceOwnershipAsync(namespaceName); - } else { - return CompletableFuture.completedFuture(null); - } - }).thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative)) + return validateTopicOperationAsync(topicName, TopicOperation.RESET_CURSOR, subName) + .thenCompose(__ -> validateGlobalNamespaceOwnershipAsync(namespaceName)) + .thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative)) .thenCompose(__ -> { // If the topic name is a partition name, no need to get partition topic metadata again if (topicName.isPartitioned()) { @@ -2206,14 +2147,9 @@ private CompletableFuture internalResetCursorForNonPartitionedTopic(String protected void internalCreateSubscription(AsyncResponse asyncResponse, String subscriptionName, MessageIdImpl messageId, boolean authoritative, boolean replicated, Map properties) { - CompletableFuture ret = validateTopicOperationAsync(topicName, TopicOperation.SUBSCRIBE, - subscriptionName); - ret.thenCompose(__ -> { - if (topicName.isGlobal()) { - return validateGlobalNamespaceOwnershipAsync(namespaceName); - } - return CompletableFuture.completedFuture(null); - }).thenAccept(__ -> { + validateTopicOperationAsync(topicName, TopicOperation.SUBSCRIBE, subscriptionName) + .thenCompose(__ -> validateGlobalNamespaceOwnershipAsync(namespaceName)) + .thenAccept(__ -> { final MessageIdImpl targetMessageId = messageId == null ? (MessageIdImpl) MessageId.latest : messageId; log.info("[{}][{}] Creating subscription {} at message id {} with properties {}", clientAppId(), topicName, subscriptionName, targetMessageId, properties); @@ -2372,13 +2308,9 @@ private void internalCreateSubscriptionForNonPartitionedTopic( protected void internalUpdateSubscriptionProperties(AsyncResponse asyncResponse, String subName, Map subscriptionProperties, boolean authoritative) { - CompletableFuture future = validateTopicOperationAsync(topicName, TopicOperation.SUBSCRIBE, subName); - future.thenCompose(__ -> { - if (topicName.isGlobal()) { - return validateGlobalNamespaceOwnershipAsync(namespaceName); - } - return CompletableFuture.completedFuture(null); - }).thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative)).thenAccept(__ -> { + validateTopicOperationAsync(topicName, TopicOperation.SUBSCRIBE, subName) + .thenCompose(__ -> validateGlobalNamespaceOwnershipAsync(namespaceName)) + .thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative)).thenAccept(__ -> { if (topicName.isPartitioned()) { internalUpdateSubscriptionPropertiesForNonPartitionedTopic(asyncResponse, subName, subscriptionProperties, authoritative); @@ -2450,13 +2382,9 @@ protected void internalUpdateSubscriptionProperties(AsyncResponse asyncResponse, protected void internalAnalyzeSubscriptionBacklog(AsyncResponse asyncResponse, String subName, Optional position, boolean authoritative) { - CompletableFuture future = validateTopicOperationAsync(topicName, TopicOperation.CONSUME, subName); - future.thenCompose(__ -> { - if (topicName.isGlobal()) { - return validateGlobalNamespaceOwnershipAsync(namespaceName); - } - return CompletableFuture.completedFuture(null); - }).thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative)) + validateTopicOperationAsync(topicName, TopicOperation.CONSUME, subName) + .thenCompose(__ -> validateGlobalNamespaceOwnershipAsync(namespaceName)) + .thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative)) .thenCompose(__ -> { if (topicName.isPartitioned()) { return CompletableFuture.completedFuture(null); @@ -2488,13 +2416,9 @@ protected void internalAnalyzeSubscriptionBacklog(AsyncResponse asyncResponse, S protected void internalGetSubscriptionProperties(AsyncResponse asyncResponse, String subName, boolean authoritative) { - CompletableFuture future = validateTopicOperationAsync(topicName, TopicOperation.CONSUME, subName); - future.thenCompose(__ -> { - if (topicName.isGlobal()) { - return validateGlobalNamespaceOwnershipAsync(namespaceName); - } - return CompletableFuture.completedFuture(null); - }).thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative)).thenAccept(__ -> { + validateTopicOperationAsync(topicName, TopicOperation.CONSUME, subName) + .thenCompose(__ -> validateGlobalNamespaceOwnershipAsync(namespaceName)) + .thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative)).thenAccept(__ -> { if (topicName.isPartitioned()) { internalGetSubscriptionPropertiesForNonPartitionedTopic(asyncResponse, subName, authoritative); @@ -2572,8 +2496,8 @@ protected void internalGetSubscriptionProperties(AsyncResponse asyncResponse, St protected void internalResetCursorOnPosition(AsyncResponse asyncResponse, String subName, boolean authoritative, MessageIdImpl messageId, boolean isExcluded, int batchIndex) { - CompletableFuture ret = validateTopicOperationAsync(topicName, TopicOperation.RESET_CURSOR, subName); - ret.thenCompose(__ -> { + validateTopicOperationAsync(topicName, TopicOperation.RESET_CURSOR, subName) + .thenCompose(__ -> { // If the topic name is a partition name, no need to get partition topic metadata again if (!topicName.isPartitioned()) { return getPartitionedTopicMetadataAsync(topicName, authoritative, false) @@ -2589,13 +2513,8 @@ protected void internalResetCursorOnPosition(AsyncResponse asyncResponse, String } else { return CompletableFuture.completedFuture(null); } - }).thenCompose(__ -> { - if (topicName.isGlobal()) { - return validateGlobalNamespaceOwnershipAsync(namespaceName); - } else { - return CompletableFuture.completedFuture(null); - } - }).thenCompose(__ -> { + }).thenCompose(__ -> validateGlobalNamespaceOwnershipAsync(namespaceName)) + .thenCompose(__ -> { log.info("[{}][{}] received reset cursor on subscription {} to position {}", clientAppId(), topicName, subName, messageId); return validateTopicOwnershipAsync(topicName, authoritative); @@ -2746,14 +2665,9 @@ private Position calculatePositionAckSet(boolean isExcluded, int batchSize, } protected CompletableFuture internalGetMessageById(long ledgerId, long entryId, boolean authoritative) { - CompletableFuture future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES); - return future.thenCompose(__ -> { - if (topicName.isGlobal()) { - return validateGlobalNamespaceOwnershipAsync(namespaceName); - } else { - return CompletableFuture.completedFuture(null); - } - }).thenCompose(__ -> { + return validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES) + .thenCompose(__ -> validateGlobalNamespaceOwnershipAsync(namespaceName)) + .thenCompose(__ -> { if (topicName.isPartitioned()) { return CompletableFuture.completedFuture(null); } else { @@ -2809,14 +2723,9 @@ public String toString() { } protected CompletableFuture internalGetMessageIdByTimestampAsync(long timestamp, boolean authoritative) { - CompletableFuture future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES); - return future.thenCompose(__ -> { - if (topicName.isGlobal()) { - return validateGlobalNamespaceOwnershipAsync(namespaceName); - } else { - return CompletableFuture.completedFuture(null); - } - }).thenCompose(__ -> { + return validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES) + .thenCompose(__ -> validateGlobalNamespaceOwnershipAsync(namespaceName)) + .thenCompose(__ -> { if (topicName.isPartitioned()) { return CompletableFuture.completedFuture(null); } else { @@ -2880,8 +2789,8 @@ private CompletableFuture findMessageIdByPublishTime(long timestamp, protected CompletableFuture internalPeekNthMessageAsync(String subName, int messagePosition, boolean authoritative) { - CompletableFuture ret = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES, subName); - return ret.thenCompose(__ -> { + return validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES, subName) + .thenCompose(__ -> { // If the topic name is a partition name, no need to get partition topic metadata again if (!topicName.isPartitioned()) { return getPartitionedTopicMetadataAsync(topicName, authoritative, false) @@ -2939,14 +2848,9 @@ protected CompletableFuture internalExamineMessageAsync(String initial boolean authoritative) { long messagePositionLocal = messagePosition < 1 ? 1 : messagePosition; String initialPositionLocal = initialPosition == null ? "latest" : initialPosition; - CompletableFuture ret = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES); - return ret.thenCompose(__ -> { - if (topicName.isGlobal()) { - return validateGlobalNamespaceOwnershipAsync(namespaceName); - } else { - return CompletableFuture.completedFuture(null); - } - }).thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative)) + return validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES) + .thenCompose(__ -> validateGlobalNamespaceOwnershipAsync(namespaceName)) + .thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative)) .thenCompose(__ -> { if (!topicName.isPartitioned()) { return getPartitionedTopicMetadataAsync(topicName, authoritative, false) @@ -3174,15 +3078,10 @@ private Response generateResponseWithEntry(Entry entry, PersistentTopic persiste } protected CompletableFuture internalGetBacklogAsync(boolean authoritative) { - CompletableFuture ret; - if (topicName.isGlobal()) { - ret = validateGlobalNamespaceOwnershipAsync(namespaceName); - } else { - ret = CompletableFuture.completedFuture(null); - } // Validate that namespace exists, throw 404 if it doesn't exist // note that we do not want to load the topic and hence skip authorization check - return ret.thenCompose(__ -> namespaceResources().getPoliciesAsync(namespaceName)) + return validateGlobalNamespaceOwnershipAsync(namespaceName) + .thenCompose(__ -> namespaceResources().getPoliciesAsync(namespaceName)) .thenCompose(__ -> { PersistentOfflineTopicStats offlineTopicStats = pulsar().getBrokerService().getOfflineTopicStat(topicName); @@ -3245,8 +3144,8 @@ protected CompletableFuture> in protected void internalGetBacklogSizeByMessageId(AsyncResponse asyncResponse, MessageIdImpl messageId, boolean authoritative) { - CompletableFuture ret = validateTopicOperationAsync(topicName, TopicOperation.GET_BACKLOG_SIZE); - ret.thenCompose(__ -> { + validateTopicOperationAsync(topicName, TopicOperation.GET_BACKLOG_SIZE) + .thenCompose(__ -> { // If the topic name is a partition name, no need to get partition topic metadata again if (!topicName.isPartitioned()) { return getPartitionedTopicMetadataAsync(topicName, authoritative, false) @@ -3262,13 +3161,8 @@ protected void internalGetBacklogSizeByMessageId(AsyncResponse asyncResponse, } else { return CompletableFuture.completedFuture(null); } - }).thenCompose(__ -> { - if (topicName.isGlobal()) { - return validateGlobalNamespaceOwnershipAsync(namespaceName); - } else { - return CompletableFuture.completedFuture(null); - } - }).thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative)) + }).thenCompose(__ -> validateGlobalNamespaceOwnershipAsync(namespaceName)) + .thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative)) .thenCompose(unused -> getTopicReferenceAsync(topicName)) .thenAccept(t -> { PersistentTopic topic = (PersistentTopic) t; @@ -3725,13 +3619,7 @@ protected CompletableFuture preValidation(boolean authoritative) { return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED, "Not allowed to set/get topic policy for a partition")); } - CompletableFuture ret; - if (topicName.isGlobal()) { - ret = validateGlobalNamespaceOwnershipAsync(namespaceName); - } else { - ret = CompletableFuture.completedFuture(null); - } - return ret + return validateGlobalNamespaceOwnershipAsync(namespaceName) .thenCompose(__ -> checkTopicExistsAsync(topicName)) .thenCompose(topicExistsInfo -> { if (!topicExistsInfo.isExists()) { @@ -3792,14 +3680,9 @@ protected CompletableFuture internalTerminateAsync(boolean authoritat "Termination of a system topic is not allowed")); } - CompletableFuture ret = validateTopicOperationAsync(topicName, TopicOperation.TERMINATE); - return ret.thenCompose(__ -> { - if (topicName.isGlobal()) { - return validateGlobalNamespaceOwnershipAsync(namespaceName); - } else { - return CompletableFuture.completedFuture(null); - } - }).thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative)) + return validateTopicOperationAsync(topicName, TopicOperation.TERMINATE) + .thenCompose(__ -> validateGlobalNamespaceOwnershipAsync(namespaceName)) + .thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative)) .thenCompose(__ -> getPartitionedTopicMetadataAsync(topicName, authoritative, false)) .thenAccept(partitionMetadata -> { if (partitionMetadata.partitions > 0) { @@ -3825,14 +3708,9 @@ protected void internalTerminatePartitionedTopic(AsyncResponse asyncResponse, bo asyncResponse.resume(new RestException(Status.METHOD_NOT_ALLOWED, msg)); return; } - CompletableFuture future = validateTopicOperationAsync(topicName, TopicOperation.TERMINATE); - future.thenCompose(__ -> { - if (topicName.isGlobal()) { - return validateGlobalNamespaceOwnershipAsync(namespaceName); - } else { - return CompletableFuture.completedFuture(null); - } - }).thenCompose(unused -> getPartitionedTopicMetadataAsync(topicName, authoritative, false)) + validateTopicOperationAsync(topicName, TopicOperation.TERMINATE) + .thenCompose(__ -> validateGlobalNamespaceOwnershipAsync(namespaceName)) + .thenCompose(unused -> getPartitionedTopicMetadataAsync(topicName, authoritative, false)) .thenAccept(partitionMetadata -> { if (partitionMetadata.partitions == 0) { String msg = "Termination of a non-partitioned topic is not allowed using partitioned-terminate" @@ -3893,15 +3771,9 @@ protected void internalTerminatePartitionedTopic(AsyncResponse asyncResponse, bo protected void internalExpireMessagesByTimestamp(AsyncResponse asyncResponse, String subName, int expireTimeInSeconds, boolean authoritative) { - CompletableFuture future = validateTopicOperationAsync(topicName, TopicOperation.EXPIRE_MESSAGES, - subName); - future.thenCompose(__ -> { - if (topicName.isGlobal()) { - return validateGlobalNamespaceOwnershipAsync(namespaceName); - } else { - return CompletableFuture.completedFuture(null); - } - }).thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative)) + validateTopicOperationAsync(topicName, TopicOperation.EXPIRE_MESSAGES, subName) + .thenCompose(__ -> validateGlobalNamespaceOwnershipAsync(namespaceName)) + .thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative)) .thenCompose(__ -> // If the topic name is a partition name, no need to get partition topic metadata again getPartitionedTopicMetadataAsync(topicName, authoritative, false) @@ -4056,8 +3928,8 @@ protected void internalExpireMessagesByPosition(AsyncResponse asyncResponse, Str asyncResponse.resume(new RestException(Status.PRECONDITION_FAILED, msg)); return; } - CompletableFuture ret = validateTopicOperationAsync(topicName, TopicOperation.EXPIRE_MESSAGES, subName); - ret.thenCompose(__ -> { + validateTopicOperationAsync(topicName, TopicOperation.EXPIRE_MESSAGES, subName) + .thenCompose(__ -> { // If the topic name is a partition name, no need to get partition topic metadata again if (!topicName.isPartitioned()) { return getPartitionedTopicMetadataAsync(topicName, authoritative, false) @@ -4072,13 +3944,8 @@ protected void internalExpireMessagesByPosition(AsyncResponse asyncResponse, Str } else { return CompletableFuture.completedFuture(null); } - }).thenCompose(__ -> { - if (topicName.isGlobal()) { - return validateGlobalNamespaceOwnershipAsync(namespaceName); - } else { - return CompletableFuture.completedFuture(null); - } - }).thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative)) + }).thenCompose(__ -> validateGlobalNamespaceOwnershipAsync(namespaceName)) + .thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative)) .thenCompose(__ -> { log.info("[{}][{}] Received expire messages on subscription {} to position {}", clientAppId(), topicName, subName, messageId); @@ -4180,14 +4047,9 @@ private CompletableFuture internalExpireMessagesNonPartitionedTopicByPosit protected void internalTriggerCompaction(AsyncResponse asyncResponse, boolean authoritative) { log.info("[{}] Trigger compaction on topic {}", clientAppId(), topicName); - CompletableFuture future = validateTopicOperationAsync(topicName, TopicOperation.COMPACT); - future.thenCompose(__ -> { - if (topicName.isGlobal()) { - return validateGlobalNamespaceOwnershipAsync(namespaceName); - } else { - return CompletableFuture.completedFuture(null); - } - }).thenAccept(__ -> { + validateTopicOperationAsync(topicName, TopicOperation.COMPACT) + .thenCompose(__ -> validateGlobalNamespaceOwnershipAsync(namespaceName)) + .thenAccept(__ -> { // If the topic name is a partition name, no need to get partition topic metadata again if (topicName.isPartitioned()) { internalTriggerCompactionNonPartitionedTopic(asyncResponse, authoritative); @@ -4957,13 +4819,6 @@ protected void internalSetReplicatedSubscriptionStatus(AsyncResponse asyncRespon return; } - // Reject the request if the topic is not global - if (!topicName.isGlobal()) { - asyncResponse.resume(new RestException(Status.METHOD_NOT_ALLOWED, - "Cannot enable/disable replicated subscriptions on non-global topics")); - return; - } - // 1.Permission to consume this topic is required // 2.Redirect the request to the peer-cluster if the local cluster is not included in the replication clusters CompletableFuture validateFuture = @@ -5097,13 +4952,6 @@ protected void internalGetReplicatedSubscriptionStatus(AsyncResponse asyncRespon return; } - // Reject the request if the topic is not global - if (!topicName.isGlobal()) { - asyncResponse.resume(new RestException(Status.METHOD_NOT_ALLOWED, - "Cannot get replicated subscriptions on non-global topics")); - return; - } - // Permission to consume this topic is required CompletableFuture validateFuture = validateTopicOperationAsync(topicName, TopicOperation.GET_REPLICATED_SUBSCRIPTION_STATUS, subName); @@ -5392,14 +5240,9 @@ protected CompletableFuture internalGetMessageIDByIndexAsync(Long ind "Invalid message index: " + index)); } int partitionIndex = topicName.getPartitionIndex(); - CompletableFuture future = validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES); - return future.thenCompose(__ -> { - if (topicName.isGlobal()) { - return validateGlobalNamespaceOwnershipAsync(namespaceName); - } else { - return CompletableFuture.completedFuture(null); - } - }).thenCompose(__ -> { + return validateTopicOperationAsync(topicName, TopicOperation.PEEK_MESSAGES) + .thenCompose(__ -> validateGlobalNamespaceOwnershipAsync(namespaceName)) + .thenCompose(__ -> { if (topicName.isPartitioned()) { return CompletableFuture.completedFuture(null); } else { diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/NonPersistentTopics.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/NonPersistentTopics.java index 9071884185d37..16cbc7104d4d8 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/NonPersistentTopics.java +++ b/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 -> { diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java index 513b16a5255f4..79a26ec56ae9f 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java +++ b/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; diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java index 2b14df499aeed..45a92c92f7ee9 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java @@ -583,7 +583,7 @@ public CompletableFuture stopReplProducers() { @Override public CompletableFuture 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 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; + }); } } } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java index da4d717e798a0..dfbb2725db44d 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java @@ -1942,7 +1942,7 @@ CompletableFuture checkPersistencePolicies() { @Override public CompletableFuture 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 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 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)) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java index 407458a1090bd..3fe51df90f72e 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java @@ -872,7 +872,7 @@ public static CompletableFuture checkLocalOrGetPeerReplicationC public static CompletableFuture checkLocalOrGetPeerReplicationCluster(PulsarService pulsarService, NamespaceName namespace, boolean allowDeletedNamespace) { - if (!namespace.isGlobal() || NamespaceService.isSLAOrHeartbeatNamespace(namespace.toString())) { + if (NamespaceService.isSLAOrHeartbeatNamespace(namespace.toString())) { return CompletableFuture.completedFuture(null); } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java index ca3f039fea41a..c40e8a84c5efc 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java +++ b/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"))); diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/naming/NamespaceName.java b/pulsar-common/src/main/java/org/apache/pulsar/common/naming/NamespaceName.java index a2e13125a9626..bda21502dfbee 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/naming/NamespaceName.java +++ b/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); } diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java b/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java index ad14edabde0bf..a1c7055a8972c 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java +++ b/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() diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/naming/NamespaceNameTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/naming/NamespaceNameTest.java index 5d2579ce4f0c8..24b86daa56200 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/naming/NamespaceNameTest.java +++ b/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"); } }