From 382145eb9ba88c20ade3bd3123068ae3d5639321 Mon Sep 17 00:00:00 2001 From: SEPURI-SAI-KRISHNA Date: Sat, 8 Aug 2026 12:04:48 +0530 Subject: [PATCH 1/2] [fix][broker] Resolve replicator remote cluster by prefix so cluster names containing a dot work Cluster names may contain dots (NamedEntity#NAMED_ENTITY_PATTERN allows "-=:." plus \w), but AbstractReplicator#getRemoteCluster recovered the cluster from a replicator cursor name by splitting on "." and taking the last segment, while getReplicatorName builds that name as .. The two were not inverses: for a cluster "remote.east" the cursor "pulsar.repl.remote.east" resolved to "east". Every admin operation addressed at such a replicator subscription failed with a 404, and PersistentTopic#removeOrphanReplicationCursors mistook the live replicator for an orphan on every topic load. Strip the known replicator prefix instead of splitting on ".", making getRemoteCluster the exact inverse of getReplicatorName. All call sites already guard with startsWith(replicatorPrefix), so the prefix is in scope at each of them. Assisted-by: Claude Code --- .../broker/admin/impl/NamespacesBase.java | 6 +- .../admin/impl/PersistentTopicsBase.java | 11 ++-- .../broker/service/AbstractReplicator.java | 24 ++++++- .../service/persistent/PersistentTopic.java | 2 +- .../service/AbstractReplicatorTest.java | 28 ++++++++ .../persistent/PersistentTopicTest.java | 64 +++++++++++++++++++ 6 files changed, 124 insertions(+), 11 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java index 911d3de2b04c2..6b935f2a62791 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java @@ -2299,9 +2299,11 @@ private CompletableFuture clearBacklogAsync(NamespaceBundle bundle, String .thenCompose(topicsInBundle -> { List> futures = new ArrayList<>(); String effectiveSubscription = subscription; + final String replicatorPrefix = pulsar().getConfiguration().getReplicatorPrefix(); if (effectiveSubscription != null - && effectiveSubscription.startsWith(pulsar().getConfiguration().getReplicatorPrefix())) { - effectiveSubscription = PersistentReplicator.getRemoteCluster(effectiveSubscription); + && effectiveSubscription.startsWith(replicatorPrefix)) { + effectiveSubscription = + PersistentReplicator.getRemoteCluster(replicatorPrefix, effectiveSubscription); } final String finalSubscription = effectiveSubscription; 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 30603404fc066..d6ca1e6913a6a 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 @@ -1979,7 +1979,8 @@ private CompletableFuture internalSkipAllMessagesForNonPartitionedTopicAsy } }; if (subName.startsWith(topic.getReplicatorPrefix())) { - String remoteCluster = PersistentReplicator.getRemoteCluster(subName); + String remoteCluster = + PersistentReplicator.getRemoteCluster(topic.getReplicatorPrefix(), subName); PersistentReplicator repl = (PersistentReplicator) topic.getPersistentReplicator(remoteCluster); if (repl == null) { @@ -2034,7 +2035,7 @@ protected void internalSkipMessages(AsyncResponse asyncResponse, String subName, getTopicNotFoundErrorMessage(topicName.toString()))); } if (subName.startsWith(topic.getReplicatorPrefix())) { - String remoteCluster = PersistentReplicator.getRemoteCluster(subName); + String remoteCluster = PersistentReplicator.getRemoteCluster(topic.getReplicatorPrefix(), subName); PersistentReplicator repl = (PersistentReplicator) topic.getPersistentReplicator(remoteCluster); if (repl == null) { @@ -4210,7 +4211,7 @@ private CompletableFuture internalExpireMessagesByTimestampForSinglePartit final MessageExpirer messageExpirer; if (subName.startsWith(topic.getReplicatorPrefix())) { - String remoteCluster = PersistentReplicator.getRemoteCluster(subName); + String remoteCluster = PersistentReplicator.getRemoteCluster(topic.getReplicatorPrefix(), subName); messageExpirer = (PersistentReplicator) topic.getPersistentReplicator(remoteCluster); } else { messageExpirer = topic.getSubscription(subName); @@ -4326,7 +4327,7 @@ private CompletableFuture internalExpireMessagesNonPartitionedTopicByPosit try { final MessageExpirer messageExpirer; if (subName.startsWith(topic.getReplicatorPrefix())) { - String remoteCluster = PersistentReplicator.getRemoteCluster(subName); + String remoteCluster = PersistentReplicator.getRemoteCluster(topic.getReplicatorPrefix(), subName); messageExpirer = (PersistentReplicator) topic.getPersistentReplicator(remoteCluster); } else { messageExpirer = topic.getSubscription(subName); @@ -4764,7 +4765,7 @@ private CompletableFuture findOrCreateSubscriptionAsync(String sub */ private PersistentReplicator getReplicatorReference(String replName, PersistentTopic topic) { try { - String remoteCluster = PersistentReplicator.getRemoteCluster(replName); + String remoteCluster = PersistentReplicator.getRemoteCluster(topic.getReplicatorPrefix(), replName); PersistentReplicator repl = (PersistentReplicator) topic.getPersistentReplicator(remoteCluster); return checkNotNull(repl); } catch (Exception e) { diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java index 4ce6684fa5dc6..6c3859bb0d73c 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java @@ -471,9 +471,27 @@ protected boolean isWritable() { return producer != null && producer.isWritable(); } - public static String getRemoteCluster(String remoteCursor) { - String[] split = remoteCursor.split("\\."); - return split[split.length - 1]; + /** + * Extract the remote cluster name from a replicator cursor/subscription name, which is the inverse of + * {@link #getReplicatorName(String, String)}: the name is {@code .}. + * + *

The known prefix is stripped instead of splitting the name on {@code '.'} and taking the last + * segment: cluster names are allowed to contain dots (see + * {@link org.apache.pulsar.common.naming.NamedEntity#NAMED_ENTITY_PATTERN}), and splitting returns only + * the part after the last dot for those — so a cluster named {@code us-east.prod} resolved to + * {@code prod}. + * + * @param replicatorPrefix the configured replicator prefix (e.g. {@code pulsar.repl}) + * @param replicatorCursorName the replicator cursor / subscription name + * @return the remote cluster name, or {@code replicatorCursorName} unchanged when it does not carry the + * prefix (the callers then fail their replicator lookup, as before) + */ + public static String getRemoteCluster(String replicatorPrefix, String replicatorCursorName) { + String prefix = replicatorPrefix + "."; + if (replicatorCursorName.startsWith(prefix)) { + return replicatorCursorName.substring(prefix.length()); + } + return replicatorCursorName; } public static String getReplicatorName(String replicatorPrefix, String cluster) { 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 3c4a404c32c12..fffa36ffd71cc 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 @@ -561,7 +561,7 @@ private CompletableFuture removeOrphanReplicationCursors() { List replicationClusters = topicPolicies.getReplicationClusters().get(); for (ManagedCursor cursor : ledger.getCursors()) { if (cursor.getName().startsWith(replicatorPrefix)) { - String remoteCluster = PersistentReplicator.getRemoteCluster(cursor.getName()); + String remoteCluster = PersistentReplicator.getRemoteCluster(replicatorPrefix, cursor.getName()); if (!replicationClusters.contains(remoteCluster)) { log.warn() .attr("remoteCluster", remoteCluster) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/AbstractReplicatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/AbstractReplicatorTest.java index 0ec9a5d0b1fad..dfecdd05d6f84 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/AbstractReplicatorTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/AbstractReplicatorTest.java @@ -133,6 +133,34 @@ public void testRetryStartProducerStoppedByTopicRemove() throws Exception { }); } + /** + * {@link AbstractReplicator#getRemoteCluster(String, String)} must be the exact inverse of + * {@link AbstractReplicator#getReplicatorName(String, String)} for every legal cluster name. Cluster names + * may contain dots ({@code NamedEntity#NAMED_ENTITY_PATTERN} allows {@code -=:.} plus word characters), so + * taking the segment after the last dot resolved {@code us-east.prod} to {@code prod}. + */ + @Test + public void testGetRemoteClusterRoundTripsClusterNamesContainingDots() { + final String replicatorPrefix = "pulsar.repl"; + for (String cluster : new String[]{"us-west", "us-east.prod", "a.b.c", "cluster:1", "r3"}) { + String cursorName = AbstractReplicator.getReplicatorName(replicatorPrefix, cluster); + Assert.assertEquals(AbstractReplicator.getRemoteCluster(replicatorPrefix, cursorName), cluster, + "remote cluster not recovered from cursor name " + cursorName); + } + } + + /** + * A prefix that is not the configured replicator prefix must not be stripped, so that callers keep failing + * their replicator lookup instead of resolving to some other cluster. + */ + @Test + public void testGetRemoteClusterLeavesNonReplicatorNamesUnchanged() { + Assert.assertEquals(AbstractReplicator.getRemoteCluster("pulsar.repl", "my-subscription"), + "my-subscription"); + Assert.assertEquals(AbstractReplicator.getRemoteCluster("pulsar.repl", "other.prefix.us-east"), + "other.prefix.us-east"); + } + private static class ReplicatorInTest extends AbstractReplicator { public ReplicatorInTest(String localCluster, Topic localTopic, String remoteCluster, String remoteTopicName, diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java index 7c9ce90180a4a..173ab0ad03781 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java @@ -94,6 +94,7 @@ import org.apache.pulsar.client.api.SubscriptionMode; import org.apache.pulsar.client.api.SubscriptionType; import org.apache.pulsar.common.naming.NamespaceBundle; +import org.apache.pulsar.common.naming.NamespaceName; import org.apache.pulsar.common.naming.TopicName; import org.apache.pulsar.common.policies.data.ClusterData; import org.apache.pulsar.common.policies.data.Policies; @@ -101,6 +102,7 @@ import org.apache.pulsar.common.policies.data.TenantInfo; import org.apache.pulsar.common.policies.data.TopicPolicies; import org.apache.pulsar.common.policies.data.TopicStats; +import org.apache.pulsar.utils.TestLogAppender; import org.awaitility.Awaitility; import org.mockito.ArgumentCaptor; import org.testng.Assert; @@ -830,6 +832,68 @@ public void testCreateTopicWithZombieReplicatorCursor(boolean topicLevelPolicy) }); } + /** + * A replicator cursor for a remote cluster whose name contains a dot must not be mistaken for an orphan. + * + *

{@code PersistentTopic#removeOrphanReplicationCursors()} used to derive the remote cluster by taking + * the cursor-name segment after the last dot, so the live cursor {@code pulsar.repl.remote.east} of the + * (legal) cluster {@code remote.east} resolved to {@code east}, which is not among the topic's replication + * clusters. The topic then tried to delete the non-existent cursor {@code pulsar.repl.east}, whose + * {@code CursorNotFoundException} failed the {@code PersistentTopic#initialize()} chain on every load of + * the topic. The live cursor survived only because the name the sweep reconstructed was wrong too. + */ + @Test + public void testReplicatorCursorOfClusterWithDotInNameIsNotTreatedAsOrphan() throws Exception { + final String namespace = "prop/ns-dotted-remote-cluster"; + final String topicName = "persistent://" + namespace + "/testDottedRemoteCluster-" + UUID.randomUUID(); + // A dot is a legal cluster-name character: NamedEntity#NAMED_ENTITY_PATTERN allows "-=:." plus \w. + final String remoteCluster = "remote.east"; + final String replicatorCursor = conf.getReplicatorPrefix() + "." + remoteCluster; + + admin.clusters().createCluster(remoteCluster, ClusterData.builder() + .serviceUrl("http://localhost:11112") + .brokerServiceUrl("pulsar://localhost:11111") + .build()); + TenantInfo tenantInfo = admin.tenants().getTenantInfo("prop"); + tenantInfo.getAllowedClusters().add(remoteCluster); + admin.tenants().updateTenant("prop", tenantInfo); + + admin.namespaces().createNamespace(namespace, Sets.newHashSet("test")); + admin.topics().createNonPartitionedTopic(topicName); + admin.topics().createSubscription(topicName, replicatorCursor, MessageId.earliest, true); + + final PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getTopic(topicName, false) + .get(10, TimeUnit.SECONDS).orElseThrow(); + + // Written straight to the namespace policies so that initialize() below reads them back synchronously, + // and to skip the admin API's remote-side validation of an intentionally unreachable cluster. + pulsar.getPulsarResources().getNamespaceResources() + .setPolicies(NamespaceName.get(namespace), policies -> { + policies.replication_clusters = Sets.newHashSet("test", remoteCluster); + return policies; + }); + + // The sweep swallows its own failure, so the warning it logs before deleting is what has to be + // asserted on: a live replicator must never reach it. + @Cleanup + final TestLogAppender logAppender = TestLogAppender.create(PersistentTopic.class); + + topic.initialize().get(30, TimeUnit.SECONDS); + + final List orphanWarnings = logAppender.getEvents().stream() + .map(event -> event.getMessage().getFormattedMessage()) + .filter(message -> message.contains("Remove the orphan replicator")) + .toList(); + assertTrue(orphanWarnings.isEmpty(), + "the live replicator of cluster " + remoteCluster + " was treated as an orphan: " + + orphanWarnings); + + final Set cursors = new HashSet<>(); + topic.getManagedLedger().getCursors().forEach(c -> cursors.add(c.getName())); + assertTrue(cursors.contains(replicatorCursor), + "the live replicator cursor was swept as an orphan, remaining cursors: " + cursors); + } + @Test public void testCheckPersistencePolicies() throws Exception { final String myNamespace = "prop/ns"; From 14774ae02d52aa066a3a07627ef157dfd063b853 Mon Sep 17 00:00:00 2001 From: SEPURI-SAI-KRISHNA Date: Thu, 10 Sep 2026 09:12:19 +0530 Subject: [PATCH 2/2] Address review: return Optional from getRemoteCluster Make AbstractReplicator#getRemoteCluster return Optional so that a single call answers both "is this a replicator name?" and "which cluster is it?", instead of every caller repeating a startsWith check that the helper then repeated internally. The prefix check is now authoritative in one place and requires the '.' separator, so a subscription that merely starts with the prefix characters, such as "pulsar.replication-state" against the prefix "pulsar.repl", is no longer treated as a replicator. The callers' separate startsWith guards are removed, and the two expire-messages call sites now reuse the same result to pick their not-found message. Assisted-by: Claude Code --- .../broker/admin/impl/NamespacesBase.java | 10 ++--- .../admin/impl/PersistentTopicsBase.java | 37 +++++++++++-------- .../broker/service/AbstractReplicator.java | 17 +++++---- .../service/persistent/PersistentTopic.java | 15 ++++---- .../service/AbstractReplicatorTest.java | 36 ++++++++++++------ 5 files changed, 65 insertions(+), 50 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java index 6b935f2a62791..a3eedb22a0ff1 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java @@ -2298,14 +2298,10 @@ private CompletableFuture clearBacklogAsync(NamespaceBundle bundle, String return pulsar().getNamespaceService().getOwnedPersistentTopicListForNamespaceBundle(bundle) .thenCompose(topicsInBundle -> { List> futures = new ArrayList<>(); - String effectiveSubscription = subscription; final String replicatorPrefix = pulsar().getConfiguration().getReplicatorPrefix(); - if (effectiveSubscription != null - && effectiveSubscription.startsWith(replicatorPrefix)) { - effectiveSubscription = - PersistentReplicator.getRemoteCluster(replicatorPrefix, effectiveSubscription); - } - final String finalSubscription = effectiveSubscription; + final String finalSubscription = subscription == null ? null + : PersistentReplicator.getRemoteCluster(replicatorPrefix, subscription) + .orElse(subscription); for (String topic : topicsInBundle) { TopicName topicName = TopicName.get(topic); 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 d6ca1e6913a6a..a25120215faca 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 @@ -76,6 +76,7 @@ import org.apache.pulsar.broker.admin.AdminResource; import org.apache.pulsar.broker.authentication.AuthenticationDataSource; import org.apache.pulsar.broker.authorization.AuthorizationService; +import org.apache.pulsar.broker.service.AbstractReplicator; import org.apache.pulsar.broker.service.AnalyzeBacklogResult; import org.apache.pulsar.broker.service.BrokerServiceException.AlreadyRunningException; import org.apache.pulsar.broker.service.BrokerServiceException.SubscriptionBusyException; @@ -1978,11 +1979,11 @@ private CompletableFuture internalSkipAllMessagesForNonPartitionedTopicAsy .log("Cleared backlog"); } }; - if (subName.startsWith(topic.getReplicatorPrefix())) { - String remoteCluster = - PersistentReplicator.getRemoteCluster(topic.getReplicatorPrefix(), subName); + Optional remoteCluster = + AbstractReplicator.getRemoteCluster(topic.getReplicatorPrefix(), subName); + if (remoteCluster.isPresent()) { PersistentReplicator repl = - (PersistentReplicator) topic.getPersistentReplicator(remoteCluster); + (PersistentReplicator) topic.getPersistentReplicator(remoteCluster.get()); if (repl == null) { asyncResponse.resume(new RestException(Status.NOT_FOUND, getSubNotFoundErrorMessage(topicName.toString(), subName))); @@ -2034,10 +2035,11 @@ protected void internalSkipMessages(AsyncResponse asyncResponse, String subName, throw new RestException(new RestException(Status.NOT_FOUND, getTopicNotFoundErrorMessage(topicName.toString()))); } - if (subName.startsWith(topic.getReplicatorPrefix())) { - String remoteCluster = PersistentReplicator.getRemoteCluster(topic.getReplicatorPrefix(), subName); + Optional remoteCluster = + AbstractReplicator.getRemoteCluster(topic.getReplicatorPrefix(), subName); + if (remoteCluster.isPresent()) { PersistentReplicator repl = - (PersistentReplicator) topic.getPersistentReplicator(remoteCluster); + (PersistentReplicator) topic.getPersistentReplicator(remoteCluster.get()); if (repl == null) { return FutureUtil.failedFuture( new RestException(Status.NOT_FOUND, "Replicator not found")); @@ -4210,14 +4212,15 @@ private CompletableFuture internalExpireMessagesByTimestampForSinglePartit PersistentTopic topic = (PersistentTopic) t; final MessageExpirer messageExpirer; - if (subName.startsWith(topic.getReplicatorPrefix())) { - String remoteCluster = PersistentReplicator.getRemoteCluster(topic.getReplicatorPrefix(), subName); - messageExpirer = (PersistentReplicator) topic.getPersistentReplicator(remoteCluster); + Optional remoteCluster = + AbstractReplicator.getRemoteCluster(topic.getReplicatorPrefix(), subName); + if (remoteCluster.isPresent()) { + messageExpirer = (PersistentReplicator) topic.getPersistentReplicator(remoteCluster.get()); } else { messageExpirer = topic.getSubscription(subName); } if (messageExpirer == null) { - final String message = subName.startsWith(topic.getReplicatorPrefix()) + final String message = remoteCluster.isPresent() ? "Replicator not found" : getSubNotFoundErrorMessage(topicName.toString(), subName); resultFuture.completeExceptionally(new RestException(Status.NOT_FOUND, message)); return; @@ -4326,14 +4329,15 @@ private CompletableFuture internalExpireMessagesNonPartitionedTopicByPosit } try { final MessageExpirer messageExpirer; - if (subName.startsWith(topic.getReplicatorPrefix())) { - String remoteCluster = PersistentReplicator.getRemoteCluster(topic.getReplicatorPrefix(), subName); - messageExpirer = (PersistentReplicator) topic.getPersistentReplicator(remoteCluster); + Optional remoteCluster = + AbstractReplicator.getRemoteCluster(topic.getReplicatorPrefix(), subName); + if (remoteCluster.isPresent()) { + messageExpirer = (PersistentReplicator) topic.getPersistentReplicator(remoteCluster.get()); } else { messageExpirer = topic.getSubscription(subName); } if (messageExpirer == null) { - final String message = (subName.startsWith(topic.getReplicatorPrefix())) + final String message = remoteCluster.isPresent() ? "Replicator not found" : getSubNotFoundErrorMessage(topicName.toString(), subName); asyncResponse.resume(new RestException(Status.NOT_FOUND, message)); return; @@ -4765,7 +4769,8 @@ private CompletableFuture findOrCreateSubscriptionAsync(String sub */ private PersistentReplicator getReplicatorReference(String replName, PersistentTopic topic) { try { - String remoteCluster = PersistentReplicator.getRemoteCluster(topic.getReplicatorPrefix(), replName); + String remoteCluster = AbstractReplicator.getRemoteCluster(topic.getReplicatorPrefix(), replName) + .orElseThrow(); PersistentReplicator repl = (PersistentReplicator) topic.getPersistentReplicator(remoteCluster); return checkNotNull(repl); } catch (Exception e) { diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java index 6c3859bb0d73c..ebe11a60f2823 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java @@ -481,17 +481,20 @@ protected boolean isWritable() { * the part after the last dot for those — so a cluster named {@code us-east.prod} resolved to * {@code prod}. * + *

This is also the authoritative test of whether a name belongs to a replicator: an empty result + * means the name is an ordinary subscription, so callers need no separate prefix check. The prefix must + * be followed by the {@code '.'} separator, so a subscription such as {@code pulsar.replication-state} + * is not mistaken for a replicator of the {@code pulsar.repl} prefix. + * * @param replicatorPrefix the configured replicator prefix (e.g. {@code pulsar.repl}) * @param replicatorCursorName the replicator cursor / subscription name - * @return the remote cluster name, or {@code replicatorCursorName} unchanged when it does not carry the - * prefix (the callers then fail their replicator lookup, as before) + * @return the remote cluster name, or empty when the name does not carry the prefix and separator */ - public static String getRemoteCluster(String replicatorPrefix, String replicatorCursorName) { + public static Optional getRemoteCluster(String replicatorPrefix, String replicatorCursorName) { String prefix = replicatorPrefix + "."; - if (replicatorCursorName.startsWith(prefix)) { - return replicatorCursorName.substring(prefix.length()); - } - return replicatorCursorName; + return replicatorCursorName.startsWith(prefix) + ? Optional.of(replicatorCursorName.substring(prefix.length())) + : Optional.empty(); } public static String getReplicatorName(String replicatorPrefix, String cluster) { 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 fffa36ffd71cc..75eeac053c065 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 @@ -560,14 +560,13 @@ private CompletableFuture removeOrphanReplicationCursors() { List> futures = new ArrayList<>(); List replicationClusters = topicPolicies.getReplicationClusters().get(); for (ManagedCursor cursor : ledger.getCursors()) { - if (cursor.getName().startsWith(replicatorPrefix)) { - String remoteCluster = PersistentReplicator.getRemoteCluster(replicatorPrefix, cursor.getName()); - if (!replicationClusters.contains(remoteCluster)) { - log.warn() - .attr("remoteCluster", remoteCluster) - .log("Remove the orphan replicator because the cluster does not exist"); - futures.add(removeReplicator(remoteCluster)); - } + Optional remoteCluster = + PersistentReplicator.getRemoteCluster(replicatorPrefix, cursor.getName()); + if (remoteCluster.isPresent() && !replicationClusters.contains(remoteCluster.get())) { + log.warn() + .attr("remoteCluster", remoteCluster.get()) + .log("Remove the orphan replicator because the cluster does not exist"); + futures.add(removeReplicator(remoteCluster.get())); } } return FutureUtil.waitForAll(futures); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/AbstractReplicatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/AbstractReplicatorTest.java index dfecdd05d6f84..29450961e06c3 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/AbstractReplicatorTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/AbstractReplicatorTest.java @@ -141,24 +141,36 @@ public void testRetryStartProducerStoppedByTopicRemove() throws Exception { */ @Test public void testGetRemoteClusterRoundTripsClusterNamesContainingDots() { - final String replicatorPrefix = "pulsar.repl"; - for (String cluster : new String[]{"us-west", "us-east.prod", "a.b.c", "cluster:1", "r3"}) { - String cursorName = AbstractReplicator.getReplicatorName(replicatorPrefix, cluster); - Assert.assertEquals(AbstractReplicator.getRemoteCluster(replicatorPrefix, cursorName), cluster, - "remote cluster not recovered from cursor name " + cursorName); + for (String replicatorPrefix : new String[]{"pulsar.repl", "repl", "my.custom.repl"}) { + for (String cluster : new String[]{"us-west", "us-east.prod", "a.b.c", "cluster:1", "r3"}) { + String cursorName = AbstractReplicator.getReplicatorName(replicatorPrefix, cluster); + Assert.assertEquals(AbstractReplicator.getRemoteCluster(replicatorPrefix, cursorName), + Optional.of(cluster), + "remote cluster not recovered from cursor name " + cursorName); + } } } /** - * A prefix that is not the configured replicator prefix must not be stripped, so that callers keep failing - * their replicator lookup instead of resolving to some other cluster. + * An empty result is what tells a caller that the name is an ordinary subscription rather than a + * replicator's, so a name must match the prefix and the {@code '.'} separator to be accepted. A + * name that merely starts with the prefix characters, such as {@code pulsar.replication-state} against + * the prefix {@code pulsar.repl}, belongs to a subscription and must not be mistaken for a replicator. */ @Test - public void testGetRemoteClusterLeavesNonReplicatorNamesUnchanged() { - Assert.assertEquals(AbstractReplicator.getRemoteCluster("pulsar.repl", "my-subscription"), - "my-subscription"); - Assert.assertEquals(AbstractReplicator.getRemoteCluster("pulsar.repl", "other.prefix.us-east"), - "other.prefix.us-east"); + public void testGetRemoteClusterIsEmptyForNamesThatAreNotReplicators() { + final String replicatorPrefix = "pulsar.repl"; + for (String notAReplicator : new String[]{ + "my-subscription", // an ordinary subscription + "other.prefix.us-east", // a different prefix entirely + "pulsar.replication-state", // starts with the prefix, but the separator does not follow + "pulsar.repl", // the bare prefix, with no separator and no cluster + "pulsar.rep", // shorter than the prefix + ""}) { + Assert.assertEquals(AbstractReplicator.getRemoteCluster(replicatorPrefix, notAReplicator), + Optional.empty(), + "name wrongly resolved to a replicator: " + notAReplicator); + } } private static class ReplicatorInTest extends AbstractReplicator {