[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

@@ -2298,12 +2298,10 @@ private CompletableFuture<Void> clearBacklogAsync(NamespaceBundle bundle, String
         return pulsar().getNamespaceService().getOwnedPersistentTopicListForNamespaceBundle(bundle)
                 .thenCompose(topicsInBundle -> {
                     List<CompletableFuture<Void>> futures = new ArrayList<>();
-                    String effectiveSubscription = subscription;
-                    if (effectiveSubscription != null
-                            && effectiveSubscription.startsWith(pulsar().getConfiguration().getReplicatorPrefix())) {
-                        effectiveSubscription = PersistentReplicator.getRemoteCluster(effectiveSubscription);
-                    }
-                    final String finalSubscription = effectiveSubscription;
+                    final String replicatorPrefix = pulsar().getConfiguration().getReplicatorPrefix();
+                    final String finalSubscription = subscription == null ? null
+                            : PersistentReplicator.getRemoteCluster(replicatorPrefix, subscription)
+                                    .orElse(subscription);
 
                     for (String topic : topicsInBundle) {
                         TopicName topicName = TopicName.get(topic);

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,10 +1979,11 @@ private CompletableFuture<Void> internalSkipAllMessagesForNonPartitionedTopicAsy
                                     .log("Cleared backlog");
                         }
                     };
-                    if (subName.startsWith(topic.getReplicatorPrefix())) {
-                        String remoteCluster = PersistentReplicator.getRemoteCluster(subName);
+                    Optional<String> 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)));
@@ -2033,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(subName);
+                 Optional<String> 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"));
@@ -4209,14 +4212,15 @@ private CompletableFuture<Void> internalExpireMessagesByTimestampForSinglePartit
                 PersistentTopic topic = (PersistentTopic) t;
 
                 final MessageExpirer messageExpirer;
-                if (subName.startsWith(topic.getReplicatorPrefix())) {
-                    String remoteCluster = PersistentReplicator.getRemoteCluster(subName);
-                    messageExpirer = (PersistentReplicator) topic.getPersistentReplicator(remoteCluster);
+                Optional<String> 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;
@@ -4325,14 +4329,15 @@ private CompletableFuture<Void> internalExpireMessagesNonPartitionedTopicByPosit
             }
             try {
                 final MessageExpirer messageExpirer;
-                if (subName.startsWith(topic.getReplicatorPrefix())) {
-                    String remoteCluster = PersistentReplicator.getRemoteCluster(subName);
-                    messageExpirer = (PersistentReplicator) topic.getPersistentReplicator(remoteCluster);
+                Optional<String> 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;
@@ -4764,7 +4769,8 @@ private CompletableFuture<Subscription> findOrCreateSubscriptionAsync(String sub
      */
     private PersistentReplicator getReplicatorReference(String replName, PersistentTopic topic) {
         try {
-            String remoteCluster = PersistentReplicator.getRemoteCluster(replName);
+            String remoteCluster = AbstractReplicator.getRemoteCluster(topic.getReplicatorPrefix(), replName)
+                    .orElseThrow();
             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,30 @@ 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}.
+     *
+     * <p>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 empty when the name does not carry the prefix and separator
+     */
+    public static Optional<String> getRemoteCluster(String replicatorPrefix, String replicatorCursorName) {
+        String prefix = replicatorPrefix + ".";
+        return replicatorCursorName.startsWith(prefix)
+                ? Optional.of(replicatorCursorName.substring(prefix.length()))
+                : Optional.empty();
     }
 
     public static String getReplicatorName(String replicatorPrefix, String cluster) {

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

@@ -560,14 +560,13 @@ private CompletableFuture<Void> removeOrphanReplicationCursors() {
         List<CompletableFuture<Void>> futures = new ArrayList<>();
         List<String> replicationClusters = topicPolicies.getReplicationClusters().get();
         for (ManagedCursor cursor : ledger.getCursors()) {
-            if (cursor.getName().startsWith(replicatorPrefix)) {
-                String remoteCluster = PersistentReplicator.getRemoteCluster(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<String> 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);

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/AbstractReplicatorTest.java

@@ -133,6 +133,46 @@ 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() {
+        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);
+            }
+        }
+    }
+
+    /**
+     * 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 <b>and</b> 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 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 {
 
         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";