You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by te...@apache.org on 2023/05/09 06:31:18 UTC
[pulsar] branch branch-2.11 updated: [fix][broker] Ignore and remove the replicator cursor when the remote cluster is absent (#19972)
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-2.11 by this push:
new 9948f89ecd7 [fix][broker] Ignore and remove the replicator cursor when the remote cluster is absent (#19972)
9948f89ecd7 is described below
commit 9948f89ecd7883078af1fdfa9244fd2dffe736ab
Author: Yunze Xu <xy...@163.com>
AuthorDate: Tue Apr 4 10:31:33 2023 +0800
[fix][broker] Ignore and remove the replicator cursor when the remote cluster is absent (#19972)
---
.../broker/service/persistent/PersistentTopic.java | 30 +++++++--
.../service/persistent/PersistentTopicTest.java | 73 +++++++++++++++++++++-
2 files changed, 96 insertions(+), 7 deletions(-)
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 56a948f3014..431be4d0811 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
@@ -1532,14 +1532,32 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
return future;
}
+ private CompletableFuture<Boolean> checkReplicationCluster(String remoteCluster) {
+ return brokerService.getPulsar().getPulsarResources().getNamespaceResources()
+ .getPoliciesAsync(TopicName.get(topic).getNamespaceObject())
+ .thenApply(optPolicies -> optPolicies.map(policies -> policies.replication_clusters)
+ .orElse(Collections.emptySet()).contains(remoteCluster)
+ || topicPolicies.getReplicationClusters().get().contains(remoteCluster));
+ }
+
protected CompletableFuture<Void> addReplicationCluster(String remoteCluster, ManagedCursor cursor,
String localCluster) {
return AbstractReplicator.validatePartitionedTopicAsync(PersistentTopic.this.getName(), brokerService)
- .thenCompose(__ -> brokerService.pulsar().getPulsarResources().getClusterResources()
- .getClusterAsync(remoteCluster)
- .thenApply(clusterData ->
- brokerService.getReplicationClient(remoteCluster, clusterData)))
+ .thenCompose(__ -> checkReplicationCluster(remoteCluster))
+ .thenCompose(clusterExists -> {
+ if (!clusterExists) {
+ log.warn("Remove the replicator because the cluster '{}' does not exist", remoteCluster);
+ return removeReplicator(remoteCluster).thenApply(__ -> null);
+ }
+ return brokerService.pulsar().getPulsarResources().getClusterResources()
+ .getClusterAsync(remoteCluster)
+ .thenApply(clusterData ->
+ brokerService.getReplicationClient(remoteCluster, clusterData));
+ })
.thenAccept(replicationClient -> {
+ if (replicationClient == null) {
+ return;
+ }
Replicator replicator = replicators.computeIfAbsent(remoteCluster, r -> {
try {
return new PersistentReplicator(PersistentTopic.this, cursor, localCluster,
@@ -1563,8 +1581,8 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
String name = PersistentReplicator.getReplicatorName(replicatorPrefix, remoteCluster);
- replicators.get(remoteCluster).disconnect().thenRun(() -> {
-
+ Optional.ofNullable(replicators.get(remoteCluster)).map(Replicator::disconnect)
+ .orElse(CompletableFuture.completedFuture(null)).thenRun(() -> {
ledger.asyncDeleteCursor(name, new DeleteCursorCallback() {
@Override
public void deleteCursorComplete(Object ctx) {
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 7a4172ca12c..186078e9b1b 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
@@ -37,7 +37,10 @@ import java.lang.reflect.Field;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
+import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
@@ -46,17 +49,30 @@ import java.util.concurrent.TimeUnit;
import com.google.common.collect.Multimap;
import com.google.common.collect.Sets;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
import lombok.Cleanup;
import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.mledger.ManagedCursor;
import org.apache.bookkeeper.mledger.ManagedLedger;
import org.apache.pulsar.broker.service.BrokerService;
import org.apache.pulsar.broker.service.BrokerTestBase;
import org.apache.pulsar.broker.stats.PrometheusMetricsTest;
import org.apache.pulsar.broker.stats.prometheus.PrometheusMetricsGenerator;
-import org.apache.pulsar.client.api.*;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.MessageListener;
+import org.apache.pulsar.client.api.MessageRoutingMode;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.common.naming.NamespaceBundle;
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.TenantInfo;
import org.apache.pulsar.common.policies.data.TopicStats;
import org.awaitility.Awaitility;
import org.junit.Assert;
@@ -402,4 +418,59 @@ public class PersistentTopicTest extends BrokerTestBase {
makeDeletedFailed.set(false);
persistentTopic.delete().get();
}
+
+ @DataProvider(name = "topicLevelPolicy")
+ public static Object[][] topicLevelPolicy() {
+ return new Object[][] { { true }, { false } };
+ }
+
+ @Test(dataProvider = "topicLevelPolicy")
+ public void testCreateTopicWithZombieReplicatorCursor(boolean topicLevelPolicy) throws Exception {
+ final String namespace = "prop/ns-abc";
+ final String topicName = "persistent://" + namespace
+ + "/testCreateTopicWithZombieReplicatorCursor" + topicLevelPolicy;
+ final String remoteCluster = "remote";
+ admin.topics().createNonPartitionedTopic(topicName);
+ admin.topics().createSubscription(topicName, conf.getReplicatorPrefix() + "." + remoteCluster,
+ MessageId.earliest, true);
+
+ 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);
+
+ if (topicLevelPolicy) {
+ admin.topics().setReplicationClusters(topicName, Collections.singletonList(remoteCluster));
+ } else {
+ admin.namespaces().setNamespaceReplicationClustersAsync(
+ namespace, Collections.singleton(remoteCluster)).get();
+ }
+
+ final PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getTopic(topicName, false)
+ .get(3, TimeUnit.SECONDS).orElse(null);
+ assertNotNull(topic);
+
+ final Supplier<Set<String>> getCursors = () -> {
+ final Set<String> cursors = new HashSet<>();
+ final Iterable<ManagedCursor> iterable = topic.getManagedLedger().getCursors();
+ iterable.forEach(c -> cursors.add(c.getName()));
+ return cursors;
+ };
+ assertEquals(getCursors.get(), Collections.singleton(conf.getReplicatorPrefix() + "." + remoteCluster));
+
+ if (topicLevelPolicy) {
+ admin.topics().setReplicationClusters(topicName, Collections.emptyList());
+ } else {
+ admin.namespaces().setNamespaceReplicationClustersAsync(namespace, Collections.emptySet()).get();
+ }
+ admin.clusters().deleteCluster(remoteCluster);
+ // Now the cluster and its related policy has been removed but the replicator cursor still exists
+
+ topic.initialize().get(3, TimeUnit.SECONDS);
+ Awaitility.await().atMost(3, TimeUnit.SECONDS)
+ .until(() -> !topic.getManagedLedger().getCursors().iterator().hasNext());
+ }
}