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";