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..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,12 +2298,10 @@ private CompletableFuture clearBacklogAsync(NamespaceBundle bundle, String return pulsar().getNamespaceService().getOwnedPersistentTopicListForNamespaceBundle(bundle) .thenCompose(topicsInBundle -> { List> 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); 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..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,10 +1979,11 @@ private CompletableFuture internalSkipAllMessagesForNonPartitionedTopicAsy .log("Cleared backlog"); } }; - if (subName.startsWith(topic.getReplicatorPrefix())) { - String remoteCluster = PersistentReplicator.getRemoteCluster(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))); @@ -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 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 internalExpireMessagesByTimestampForSinglePartit PersistentTopic topic = (PersistentTopic) t; final MessageExpirer messageExpirer; - if (subName.startsWith(topic.getReplicatorPrefix())) { - String remoteCluster = PersistentReplicator.getRemoteCluster(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; @@ -4325,14 +4329,15 @@ private CompletableFuture internalExpireMessagesNonPartitionedTopicByPosit } try { final MessageExpirer messageExpirer; - if (subName.startsWith(topic.getReplicatorPrefix())) { - String remoteCluster = PersistentReplicator.getRemoteCluster(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; @@ -4764,7 +4769,8 @@ private CompletableFuture 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) { 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..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 @@ -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 .}. + * + *

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}. + * + *

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 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) { 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..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(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 0ec9a5d0b1fad..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 @@ -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 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 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, 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";