[fix][broker] Resolve replicator remote cluster by prefix so cluster names containing a dot work

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

@@ -2299,9 +2299,11 @@ private CompletableFuture<Void> clearBacklogAsync(NamespaceBundle bundle, String
                 .thenCompose(topicsInBundle -> {
                     List<CompletableFuture<Void>> 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;
 

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

@@ -1979,7 +1979,8 @@ private CompletableFuture<Void> 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<Void> 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<Void> 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<Subscription> 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) {

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 <replicatorPrefix>.<remoteCluster>}.
+     *
+     * <p>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) {

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

@@ -561,7 +561,7 @@ private CompletableFuture<Void> removeOrphanReplicationCursors() {
         List<String> 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)

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,

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java

@@ -94,13 +94,15 @@
 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;
 import org.apache.pulsar.common.policies.data.RetentionPolicies;
 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.
+     *
+     * <p>{@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<String> 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<String> 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";