From 382145eb9ba88c20ade3bd3123068ae3d5639321 Mon Sep 17 00:00:00 2001 From: SEPURI-SAI-KRISHNA Date: Sat, 8 Aug 2026 12:04:48 +0530 Subject: [PATCH] [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";