You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2024/06/07 18:16:40 UTC
(pulsar) branch branch-3.3 updated (6e3adc81755 -> 9677d569a69)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 6e3adc81755 Bump version to 3.3.1-SNAPSHOT
new 6626390cebd [fix] [broker] fix deadlock when disable topic level Geo-Replication (#22738)
new 9677d569a69 [fix] [broker] fix topic partitions was expanded even if disabled topic level replication (#22769)
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../broker/admin/impl/PersistentTopicsBase.java | 17 +-
.../broker/service/OneWayReplicatorTest.java | 124 +++++++++++++-
.../broker/service/OneWayReplicatorTestBase.java | 185 +++++++++++++++++----
.../service/OneWayReplicatorUsingGlobalZKTest.java | 107 ++++++++++++
4 files changed, 399 insertions(+), 34 deletions(-)
create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalZKTest.java
(pulsar) 02/02: [fix] [broker] fix topic partitions was expanded even if disabled topic level replication (#22769)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 9677d569a69d6bf5afe1822044416474d6ab6aa0
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Tue May 28 11:13:12 2024 +0800
[fix] [broker] fix topic partitions was expanded even if disabled topic level replication (#22769)
(cherry picked from commit 55ad4b22ba2e94029c2e1c01b67b22cb237e5ecc)
---
.../broker/admin/impl/PersistentTopicsBase.java | 11 +++-
.../broker/service/OneWayReplicatorTest.java | 65 +++++++++++++++++++++-
.../broker/service/OneWayReplicatorTestBase.java | 13 ++++-
.../service/OneWayReplicatorUsingGlobalZKTest.java | 10 ++++
4 files changed, 93 insertions(+), 6 deletions(-)
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 9c2746a0293..eec6370cec6 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
@@ -35,6 +35,7 @@ import java.util.ArrayList;
import java.util.Base64;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -452,7 +453,14 @@ public class PersistentTopicsBase extends AdminResource {
if (!policies.isPresent()) {
return CompletableFuture.completedFuture(null);
}
- final Set<String> replicationClusters = policies.get().replication_clusters;
+ // Combine namespace level policies and topic level policies.
+ Set<String> replicationClusters = policies.get().replication_clusters;
+ TopicPolicies topicPolicies =
+ pulsarService.getTopicPoliciesService().getTopicPoliciesIfExists(topicName);
+ if (topicPolicies != null) {
+ replicationClusters = new HashSet<>(topicPolicies.getReplicationClusters());
+ }
+ // Do check replicated clusters.
if (replicationClusters.size() == 0) {
return CompletableFuture.completedFuture(null);
}
@@ -468,6 +476,7 @@ public class PersistentTopicsBase extends AdminResource {
// The replication clusters just has the current cluster itself.
return CompletableFuture.completedFuture(null);
}
+ // Do sync operation to other clusters.
List<CompletableFuture<Void>> futures = replicationClusters.stream()
.map(replicationCluster -> admin.clusters().getClusterAsync(replicationCluster)
.thenCompose(clusterData -> pulsarService.getBrokerService()
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java
index a5f1339e95f..3dcd787a0cd 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java
@@ -495,8 +495,17 @@ public class OneWayReplicatorTest extends OneWayReplicatorTestBase {
admin2.topics().createPartitionedTopic(topicName, 2);
admin1.topics().setReplicationClusters(topicName, Arrays.asList(cluster1, cluster2));
// Check the partitioned topic has been created at the remote cluster.
- PartitionedTopicMetadata topicMetadata2 = admin2.topics().getPartitionedTopicMetadata(topicName);
- assertEquals(topicMetadata2.partitions, 2);
+ Awaitility.await().untilAsserted(() -> {
+ PartitionedTopicMetadata topicMetadata2 = admin2.topics().getPartitionedTopicMetadata(topicName);
+ assertEquals(topicMetadata2.partitions, 2);
+ });
+
+ // Expand partitions
+ admin2.topics().updatePartitionedTopic(topicName, 3);
+ Awaitility.await().untilAsserted(() -> {
+ PartitionedTopicMetadata topicMetadata2 = admin2.topics().getPartitionedTopicMetadata(topicName);
+ assertEquals(topicMetadata2.partitions, 3);
+ });
// cleanup.
admin1.topics().setReplicationClusters(topicName, Arrays.asList(cluster1));
waitReplicatorStopped(partition0);
@@ -748,4 +757,56 @@ public class OneWayReplicatorTest extends OneWayReplicatorTestBase {
.persistentTopicExists(TopicName.get(topicName).getPartition(1)).join());
}
}
+
+ @Test
+ public void testNoExpandTopicPartitionsWhenDisableTopicLevelReplication() throws Exception {
+ final String topicName = BrokerTestUtil.newUniqueName("persistent://" + replicatedNamespace + "/tp_");
+ admin1.topics().createPartitionedTopic(topicName, 2);
+
+ // Verify replicator works.
+ verifyReplicationWorks(topicName);
+
+ // Disable topic level replication.
+ setTopicLevelClusters(topicName, Arrays.asList(cluster1), admin1, pulsar1);
+ setTopicLevelClusters(topicName, Arrays.asList(cluster2), admin2, pulsar2);
+
+ // Expand topic.
+ admin1.topics().updatePartitionedTopic(topicName, 3);
+ assertEquals(admin1.topics().getPartitionedTopicMetadata(topicName).partitions, 3);
+
+ // Wait for async tasks that were triggered by expanding topic partitions.
+ Thread.sleep(3 * 1000);
+
+
+ // Verify: the topics on the remote cluster did not been expanded.
+ assertEquals(admin2.topics().getPartitionedTopicMetadata(topicName).partitions, 2);
+
+ cleanupTopics(() -> {
+ admin1.topics().deletePartitionedTopic(topicName, false);
+ admin2.topics().deletePartitionedTopic(topicName, false);
+ });
+ }
+
+ @Test
+ public void testExpandTopicPartitionsOnNamespaceLevelReplication() throws Exception {
+ final String topicName = BrokerTestUtil.newUniqueName("persistent://" + replicatedNamespace + "/tp_");
+ admin1.topics().createPartitionedTopic(topicName, 2);
+
+ // Verify replicator works.
+ verifyReplicationWorks(topicName);
+
+ // Expand topic.
+ admin1.topics().updatePartitionedTopic(topicName, 3);
+ assertEquals(admin1.topics().getPartitionedTopicMetadata(topicName).partitions, 3);
+
+ // Verify: the topics on the remote cluster will be expanded.
+ Awaitility.await().untilAsserted(() -> {
+ assertEquals(admin2.topics().getPartitionedTopicMetadata(topicName).partitions, 3);
+ });
+
+ cleanupTopics(() -> {
+ admin1.topics().deletePartitionedTopic(topicName, false);
+ admin2.topics().deletePartitionedTopic(topicName, false);
+ });
+ }
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTestBase.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTestBase.java
index 6a84432890c..7372b2e4784 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTestBase.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTestBase.java
@@ -196,9 +196,16 @@ public abstract class OneWayReplicatorTestBase extends TestRetrySupport {
}
protected void waitChangeEventsInit(String namespace) {
- PersistentTopic topic = (PersistentTopic) pulsar1.getBrokerService()
- .getTopic(namespace + "/" + SystemTopicNames.NAMESPACE_EVENTS_LOCAL_NAME, false)
- .join().get();
+ CompletableFuture<Optional<Topic>> future = pulsar1.getBrokerService()
+ .getTopic(namespace + "/" + SystemTopicNames.NAMESPACE_EVENTS_LOCAL_NAME, false);
+ if (future == null) {
+ return;
+ }
+ Optional<Topic> optional = future.join();
+ if (!optional.isPresent()) {
+ return;
+ }
+ PersistentTopic topic = (PersistentTopic) optional.get();
Awaitility.await().atMost(Duration.ofSeconds(180)).untilAsserted(() -> {
TopicStatsImpl topicStats = topic.getStats(true, false, false);
topicStats.getSubscriptions().entrySet().forEach(entry -> {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalZKTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalZKTest.java
index d827235bc32..b4747a8bd0e 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalZKTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalZKTest.java
@@ -94,4 +94,14 @@ public class OneWayReplicatorUsingGlobalZKTest extends OneWayReplicatorTest {
public void testDeletePartitionedTopic() throws Exception {
super.testDeletePartitionedTopic();
}
+
+ @Test(enabled = false)
+ public void testNoExpandTopicPartitionsWhenDisableTopicLevelReplication() throws Exception {
+ super.testNoExpandTopicPartitionsWhenDisableTopicLevelReplication();
+ }
+
+ @Test(enabled = false)
+ public void testExpandTopicPartitionsOnNamespaceLevelReplication() throws Exception {
+ super.testExpandTopicPartitionsOnNamespaceLevelReplication();
+ }
}
(pulsar) 01/02: [fix] [broker] fix deadlock when disable topic level Geo-Replication (#22738)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 6626390cebd381e8be324a06c8d5215e2ee072c8
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Sun May 19 11:28:21 2024 +0800
[fix] [broker] fix deadlock when disable topic level Geo-Replication (#22738)
(cherry picked from commit 6372b9c8f5448757cdb56dc70aaf0ebb52a1063e)
---
.../broker/admin/impl/PersistentTopicsBase.java | 6 +-
.../broker/service/OneWayReplicatorTest.java | 59 +++++++
.../broker/service/OneWayReplicatorTestBase.java | 172 +++++++++++++++++----
.../service/OneWayReplicatorUsingGlobalZKTest.java | 97 ++++++++++++
4 files changed, 306 insertions(+), 28 deletions(-)
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 3461eb2d140..9c2746a0293 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
@@ -3274,12 +3274,14 @@ public class PersistentTopicsBase extends AdminResource {
}
Set<String> replicationClusters = Sets.newHashSet(clusterIds);
return validatePoliciesReadOnlyAccessAsync()
- .thenCompose(__ -> {
+ .thenAccept(__ -> {
if (replicationClusters.contains("global")) {
throw new RestException(Status.PRECONDITION_FAILED,
"Cannot specify global in the list of replication clusters");
}
- Set<String> clusters = clusters();
+ })
+ .thenCompose(__ -> clustersAsync())
+ .thenCompose(clusters -> {
List<CompletableFuture<Void>> futures = new ArrayList<>(replicationClusters.size());
for (String clusterId : replicationClusters) {
if (!clusters.contains(clusterId)) {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java
index fae72e8eac2..a5f1339e95f 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java
@@ -689,4 +689,63 @@ public class OneWayReplicatorTest extends OneWayReplicatorTestBase {
admin2.topics().delete(topicName);
});
}
+
+ @Test
+ public void testDeleteNonPartitionedTopic() throws Exception {
+ final String topicName = BrokerTestUtil.newUniqueName("persistent://" + replicatedNamespace + "/tp_");
+ admin1.topics().createNonPartitionedTopic(topicName);
+
+ // Verify replicator works.
+ verifyReplicationWorks(topicName);
+
+ // Disable replication.
+ setTopicLevelClusters(topicName, Arrays.asList(cluster1), admin1, pulsar1);
+ setTopicLevelClusters(topicName, Arrays.asList(cluster2), admin2, pulsar2);
+
+ // Delete topic.
+ admin1.topics().delete(topicName);
+ admin2.topics().delete(topicName);
+
+ // Verify the topic was deleted.
+ assertFalse(pulsar1.getPulsarResources().getTopicResources()
+ .persistentTopicExists(TopicName.get(topicName)).join());
+ assertFalse(pulsar2.getPulsarResources().getTopicResources()
+ .persistentTopicExists(TopicName.get(topicName)).join());
+ }
+
+ @Test
+ public void testDeletePartitionedTopic() throws Exception {
+ final String topicName = BrokerTestUtil.newUniqueName("persistent://" + replicatedNamespace + "/tp_");
+ admin1.topics().createPartitionedTopic(topicName, 2);
+
+ // Verify replicator works.
+ verifyReplicationWorks(topicName);
+
+ // Disable replication.
+ setTopicLevelClusters(topicName, Arrays.asList(cluster1), admin1, pulsar1);
+ setTopicLevelClusters(topicName, Arrays.asList(cluster2), admin2, pulsar2);
+
+ // Delete topic.
+ admin1.topics().deletePartitionedTopic(topicName);
+ if (!usingGlobalZK) {
+ admin2.topics().deletePartitionedTopic(topicName);
+ }
+
+ // Verify the topic was deleted.
+ assertFalse(pulsar1.getPulsarResources().getNamespaceResources().getPartitionedTopicResources()
+ .partitionedTopicExists(TopicName.get(topicName)));
+ assertFalse(pulsar2.getPulsarResources().getNamespaceResources().getPartitionedTopicResources()
+ .partitionedTopicExists(TopicName.get(topicName)));
+ if (!usingGlobalZK) {
+ // So far, the topic partitions on the remote cluster are needed to delete manually when using global ZK.
+ assertFalse(pulsar1.getPulsarResources().getTopicResources()
+ .persistentTopicExists(TopicName.get(topicName).getPartition(0)).join());
+ assertFalse(pulsar2.getPulsarResources().getTopicResources()
+ .persistentTopicExists(TopicName.get(topicName).getPartition(0)).join());
+ assertFalse(pulsar1.getPulsarResources().getTopicResources()
+ .persistentTopicExists(TopicName.get(topicName).getPartition(1)).join());
+ assertFalse(pulsar2.getPulsarResources().getTopicResources()
+ .persistentTopicExists(TopicName.get(topicName).getPartition(1)).join());
+ }
+ }
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTestBase.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTestBase.java
index 317e43306e3..6a84432890c 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTestBase.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTestBase.java
@@ -21,21 +21,33 @@ package org.apache.pulsar.broker.service;
import static org.apache.pulsar.compaction.Compactor.COMPACTION_SUBSCRIPTION;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.assertEquals;
import com.google.common.collect.Sets;
import java.net.URL;
import java.time.Duration;
+import java.util.Arrays;
import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.api.ClientBuilder;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.common.naming.SystemTopicNames;
+import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.ClusterData;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
+import org.apache.pulsar.common.policies.data.TopicPolicies;
import org.apache.pulsar.common.policies.data.TopicType;
import org.apache.pulsar.common.policies.data.stats.TopicStatsImpl;
import org.apache.pulsar.tests.TestRetrySupport;
@@ -52,6 +64,9 @@ public abstract class OneWayReplicatorTestBase extends TestRetrySupport {
protected final String nonReplicatedNamespace = defaultTenant + "/ns1";
protected final String cluster1 = "r1";
+
+ protected boolean usingGlobalZK = false;
+
protected URL url1;
protected URL urlTls1;
protected ServiceConfiguration config1 = new ServiceConfiguration();
@@ -77,8 +92,12 @@ public abstract class OneWayReplicatorTestBase extends TestRetrySupport {
// Start ZK.
brokerConfigZk1 = new ZookeeperServerTest(0);
brokerConfigZk1.start();
- brokerConfigZk2 = new ZookeeperServerTest(0);
- brokerConfigZk2.start();
+ if (usingGlobalZK) {
+ brokerConfigZk2 = brokerConfigZk1;
+ } else {
+ brokerConfigZk2 = new ZookeeperServerTest(0);
+ brokerConfigZk2.start();
+ }
// Start BK.
bkEnsemble1 = new LocalBookkeeperEnsemble(3, 0, () -> 0);
@@ -132,30 +151,32 @@ public abstract class OneWayReplicatorTestBase extends TestRetrySupport {
.brokerServiceUrlTls(pulsar2.getBrokerServiceUrlTls())
.brokerClientTlsEnabled(false)
.build());
- admin2.clusters().createCluster(cluster1, ClusterData.builder()
- .serviceUrl(url1.toString())
- .serviceUrlTls(urlTls1.toString())
- .brokerServiceUrl(pulsar1.getBrokerServiceUrl())
- .brokerServiceUrlTls(pulsar1.getBrokerServiceUrlTls())
- .brokerClientTlsEnabled(false)
- .build());
- admin2.clusters().createCluster(cluster2, ClusterData.builder()
- .serviceUrl(url2.toString())
- .serviceUrlTls(urlTls2.toString())
- .brokerServiceUrl(pulsar2.getBrokerServiceUrl())
- .brokerServiceUrlTls(pulsar2.getBrokerServiceUrlTls())
- .brokerClientTlsEnabled(false)
- .build());
-
admin1.tenants().createTenant(defaultTenant, new TenantInfoImpl(Collections.emptySet(),
Sets.newHashSet(cluster1, cluster2)));
- admin2.tenants().createTenant(defaultTenant, new TenantInfoImpl(Collections.emptySet(),
- Sets.newHashSet(cluster1, cluster2)));
-
admin1.namespaces().createNamespace(replicatedNamespace, Sets.newHashSet(cluster1, cluster2));
- admin2.namespaces().createNamespace(replicatedNamespace);
admin1.namespaces().createNamespace(nonReplicatedNamespace);
- admin2.namespaces().createNamespace(nonReplicatedNamespace);
+
+ if (!usingGlobalZK) {
+ admin2.clusters().createCluster(cluster1, ClusterData.builder()
+ .serviceUrl(url1.toString())
+ .serviceUrlTls(urlTls1.toString())
+ .brokerServiceUrl(pulsar1.getBrokerServiceUrl())
+ .brokerServiceUrlTls(pulsar1.getBrokerServiceUrlTls())
+ .brokerClientTlsEnabled(false)
+ .build());
+ admin2.clusters().createCluster(cluster2, ClusterData.builder()
+ .serviceUrl(url2.toString())
+ .serviceUrlTls(urlTls2.toString())
+ .brokerServiceUrl(pulsar2.getBrokerServiceUrl())
+ .brokerServiceUrlTls(pulsar2.getBrokerServiceUrlTls())
+ .brokerClientTlsEnabled(false)
+ .build());
+ admin2.tenants().createTenant(defaultTenant, new TenantInfoImpl(Collections.emptySet(),
+ Sets.newHashSet(cluster1, cluster2)));
+ admin2.namespaces().createNamespace(replicatedNamespace);
+ admin2.namespaces().createNamespace(nonReplicatedNamespace);
+ }
+
}
protected void cleanupTopics(CleanupTopicAction cleanupTopicAction) throws Exception {
@@ -163,6 +184,9 @@ public abstract class OneWayReplicatorTestBase extends TestRetrySupport {
}
protected void cleanupTopics(String namespace, CleanupTopicAction cleanupTopicAction) throws Exception {
+ if (usingGlobalZK) {
+ throw new IllegalArgumentException("The method cleanupTopics does not support for global ZK");
+ }
waitChangeEventsInit(namespace);
admin1.namespaces().setNamespaceReplicationClusters(namespace, Collections.singleton(cluster1));
admin1.namespaces().unload(namespace);
@@ -242,11 +266,15 @@ public abstract class OneWayReplicatorTestBase extends TestRetrySupport {
// delete namespaces.
waitChangeEventsInit(replicatedNamespace);
admin1.namespaces().setNamespaceReplicationClusters(replicatedNamespace, Sets.newHashSet(cluster1));
+ if (!usingGlobalZK) {
+ admin2.namespaces().setNamespaceReplicationClusters(replicatedNamespace, Sets.newHashSet(cluster2));
+ }
admin1.namespaces().deleteNamespace(replicatedNamespace);
- admin2.namespaces().setNamespaceReplicationClusters(replicatedNamespace, Sets.newHashSet(cluster2));
- admin2.namespaces().deleteNamespace(replicatedNamespace);
admin1.namespaces().deleteNamespace(nonReplicatedNamespace);
- admin2.namespaces().deleteNamespace(nonReplicatedNamespace);
+ if (!usingGlobalZK) {
+ admin2.namespaces().deleteNamespace(replicatedNamespace);
+ admin2.namespaces().deleteNamespace(nonReplicatedNamespace);
+ }
// shutdown.
markCurrentSetupNumberCleaned();
@@ -291,7 +319,7 @@ public abstract class OneWayReplicatorTestBase extends TestRetrySupport {
brokerConfigZk1.stop();
brokerConfigZk1 = null;
}
- if (brokerConfigZk2 != null) {
+ if (!usingGlobalZK && brokerConfigZk2 != null) {
brokerConfigZk2.stop();
brokerConfigZk2 = null;
}
@@ -313,4 +341,96 @@ public abstract class OneWayReplicatorTestBase extends TestRetrySupport {
protected PulsarClient initClient(ClientBuilder clientBuilder) throws Exception {
return clientBuilder.build();
}
+
+ protected void verifyReplicationWorks(String topic) throws Exception {
+ final String subscription = "__subscribe_1";
+ final String msgValue = "__msg1";
+ Producer<String> producer1 = client1.newProducer(Schema.STRING).topic(topic).create();
+ Consumer<String> consumer2 = client2.newConsumer(Schema.STRING).topic(topic).isAckReceiptEnabled(true)
+ .subscriptionName(subscription).subscribe();
+ producer1.newMessage().value(msgValue).send();
+ pulsar1.getBrokerService().checkReplicationPolicies();
+ assertEquals(consumer2.receive(10, TimeUnit.SECONDS).getValue(), msgValue);
+ consumer2.unsubscribe();
+ producer1.close();
+ }
+
+ protected void setTopicLevelClusters(String topic, List<String> clusters, PulsarAdmin admin,
+ PulsarService pulsar) throws Exception {
+ Set<String> expected = new HashSet<>(clusters);
+ TopicName topicName = TopicName.get(TopicName.get(topic).getPartitionedTopicName());
+ int partitions = ensurePartitionsAreSame(topic);
+ admin.topics().setReplicationClusters(topic, clusters);
+ Awaitility.await().untilAsserted(() -> {
+ TopicPolicies policies = pulsar.getTopicPoliciesService().getTopicPolicies(topicName);
+ assertEquals(new HashSet<>(policies.getReplicationClusters()), expected);
+ if (partitions == 0) {
+ checkNonPartitionedTopicLevelClusters(topicName.toString(), clusters, admin, pulsar.getBrokerService());
+ } else {
+ for (int i = 0; i < partitions; i++) {
+ checkNonPartitionedTopicLevelClusters(topicName.getPartition(i).toString(), clusters, admin,
+ pulsar.getBrokerService());
+ }
+ }
+ });
+ }
+
+ protected void checkNonPartitionedTopicLevelClusters(String topic, List<String> clusters, PulsarAdmin admin,
+ BrokerService broker) throws Exception {
+ CompletableFuture<Optional<Topic>> future = broker.getTopic(topic, false);
+ if (future == null) {
+ return;
+ }
+ Optional<Topic> optional = future.join();
+ if (optional == null || !optional.isPresent()) {
+ return;
+ }
+ PersistentTopic persistentTopic = (PersistentTopic) optional.get();
+ Set<String> expected = new HashSet<>(clusters);
+ Set<String> act = new HashSet<>(persistentTopic.getTopicPolicies().get().getReplicationClusters());
+ assertEquals(act, expected);
+ }
+
+ protected int ensurePartitionsAreSame(String topic) throws Exception {
+ TopicName topicName = TopicName.get(TopicName.get(topic).getPartitionedTopicName());
+ boolean isPartitionedTopic1 = pulsar1.getPulsarResources().getNamespaceResources()
+ .getPartitionedTopicResources().partitionedTopicExists(topicName);
+ boolean isPartitionedTopic2 = pulsar2.getPulsarResources().getNamespaceResources()
+ .getPartitionedTopicResources().partitionedTopicExists(topicName);
+ if (isPartitionedTopic1 != isPartitionedTopic2) {
+ throw new IllegalArgumentException(String.format("Can not delete topic."
+ + " isPartitionedTopic1: %s, isPartitionedTopic2: %s",
+ isPartitionedTopic1, isPartitionedTopic2));
+ }
+ if (!isPartitionedTopic1) {
+ return 0;
+ }
+ int partitions1 = pulsar1.getPulsarResources().getNamespaceResources()
+ .getPartitionedTopicResources().getPartitionedTopicMetadataAsync(topicName).join().get().partitions;
+ int partitions2 = pulsar2.getPulsarResources().getNamespaceResources()
+ .getPartitionedTopicResources().getPartitionedTopicMetadataAsync(topicName).join().get().partitions;
+ if (partitions1 != partitions2) {
+ throw new IllegalArgumentException(String.format("Can not delete topic."
+ + " partitions1: %s, partitions2: %s",
+ partitions1, partitions2));
+ }
+ return partitions1;
+ }
+
+ protected void deleteTopicAfterDisableTopicLevelReplication(String topic) throws Exception {
+ setTopicLevelClusters(topic, Arrays.asList(cluster1), admin1, pulsar1);
+ setTopicLevelClusters(topic, Arrays.asList(cluster1), admin2, pulsar2);
+ admin2.topics().setReplicationClusters(topic, Arrays.asList(cluster2));
+
+ int partitions = ensurePartitionsAreSame(topic);
+
+ TopicName topicName = TopicName.get(TopicName.get(topic).getPartitionedTopicName());
+ if (partitions != 0) {
+ admin1.topics().deletePartitionedTopic(topicName.toString());
+ admin2.topics().deletePartitionedTopic(topicName.toString());
+ } else {
+ admin1.topics().delete(topicName.toString());
+ admin2.topics().delete(topicName.toString());
+ }
+ }
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalZKTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalZKTest.java
new file mode 100644
index 00000000000..d827235bc32
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalZKTest.java
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service;
+
+import lombok.extern.slf4j.Slf4j;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+@Slf4j
+@Test(groups = "broker")
+public class OneWayReplicatorUsingGlobalZKTest extends OneWayReplicatorTest {
+
+ @Override
+ @BeforeClass(alwaysRun = true, timeOut = 300000)
+ public void setup() throws Exception {
+ super.usingGlobalZK = true;
+ super.setup();
+ }
+
+ @Override
+ @AfterClass(alwaysRun = true, timeOut = 300000)
+ public void cleanup() throws Exception {
+ super.cleanup();
+ }
+
+ @Test(enabled = false)
+ public void testReplicatorProducerStatInTopic() throws Exception {
+ super.testReplicatorProducerStatInTopic();
+ }
+
+ @Test(enabled = false)
+ public void testCreateRemoteConsumerFirst() throws Exception {
+ super.testReplicatorProducerStatInTopic();
+ }
+
+ @Test(enabled = false)
+ public void testTopicCloseWhenInternalProducerCloseErrorOnce() throws Exception {
+ super.testReplicatorProducerStatInTopic();
+ }
+
+ @Test(enabled = false)
+ public void testConcurrencyOfUnloadBundleAndRecreateProducer() throws Exception {
+ super.testConcurrencyOfUnloadBundleAndRecreateProducer();
+ }
+
+ @Test(enabled = false)
+ public void testPartitionedTopicLevelReplication() throws Exception {
+ super.testPartitionedTopicLevelReplication();
+ }
+
+ @Test(enabled = false)
+ public void testPartitionedTopicLevelReplicationRemoteTopicExist() throws Exception {
+ super.testPartitionedTopicLevelReplicationRemoteTopicExist();
+ }
+
+ @Test(enabled = false)
+ public void testPartitionedTopicLevelReplicationRemoteConflictTopicExist() throws Exception {
+ super.testPartitionedTopicLevelReplicationRemoteConflictTopicExist();
+ }
+
+ @Test(enabled = false)
+ public void testConcurrencyOfUnloadBundleAndRecreateProducer2() throws Exception {
+ super.testConcurrencyOfUnloadBundleAndRecreateProducer2();
+ }
+
+ @Test(enabled = false)
+ public void testUnFenceTopicToReuse() throws Exception {
+ super.testUnFenceTopicToReuse();
+ }
+
+ @Test
+ public void testDeleteNonPartitionedTopic() throws Exception {
+ super.testDeleteNonPartitionedTopic();
+ }
+
+ @Test
+ public void testDeletePartitionedTopic() throws Exception {
+ super.testDeletePartitionedTopic();
+ }
+}