You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by yu...@apache.org on 2024/05/20 03:24:12 UTC
(pulsar) branch branch-3.0 updated (71640f696fd -> 1a8ba1356a1)
This is an automated email from the ASF dual-hosted git repository.
yubiao pushed a change to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 71640f696fd [improve][admin] Check if the topic existed before the permission operations (#22742)
new b11ab69898f [improve] [test] Add a test to guarantee the TNX topics will not be replicated (#22721)
new 1a8ba1356a1 [fix] [broker] fix deadlock when disable topic level Geo-Replication (#22738)
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 | 5 +-
.../broker/service/OneWayReplicatorTest.java | 68 +++++-
.../broker/service/OneWayReplicatorTestBase.java | 216 ++++++++++++++---
.../service/OneWayReplicatorUsingGlobalZKTest.java | 97 ++++++++
.../pulsar/broker/service/ReplicationTxnTest.java | 262 +++++++++++++++++++++
5 files changed, 602 insertions(+), 46 deletions(-)
create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalZKTest.java
create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicationTxnTest.java
(pulsar) 02/02: [fix] [broker] fix deadlock when disable topic level Geo-Replication (#22738)
Posted by yu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yubiao pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 1a8ba1356a18e6794a411239e411202c77965c4d
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 | 5 +-
.../broker/service/OneWayReplicatorTest.java | 59 +++++++
.../broker/service/OneWayReplicatorTestBase.java | 172 +++++++++++++++++----
.../service/OneWayReplicatorUsingGlobalZKTest.java | 97 ++++++++++++
4 files changed, 305 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 978111d7187..057302345d1 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
@@ -3445,11 +3445,12 @@ public class PersistentTopicsBase extends AdminResource {
Set<String> replicationClusters = Sets.newHashSet(clusterIds);
return validateTopicPolicyOperationAsync(topicName, PolicyName.REPLICATION, PolicyOperation.WRITE)
.thenCompose(__ -> validatePoliciesReadOnlyAccessAsync())
- .thenCompose(__ -> {
+ .thenAccept(__ -> {
if (CollectionUtils.isEmpty(clusterIds)) {
throw new RestException(Status.PRECONDITION_FAILED, "ClusterIds should not be null or empty");
}
- 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 30c7a41e28d..771981d7306 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
@@ -658,4 +658,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();
+ }
+}
(pulsar) 01/02: [improve] [test] Add a test to guarantee the TNX topics will not be replicated (#22721)
Posted by yu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yubiao pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit b11ab69898fe6091cb5c196bc2418924b8ebf7a7
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Fri May 17 01:18:49 2024 +0800
[improve] [test] Add a test to guarantee the TNX topics will not be replicated (#22721)
(cherry picked from commit 4e132d3f2919767cef6fa935a59721937cb668e7)
---
.../broker/service/OneWayReplicatorTest.java | 9 -
.../broker/service/OneWayReplicatorTestBase.java | 44 +++-
.../pulsar/broker/service/ReplicationTxnTest.java | 262 +++++++++++++++++++++
3 files changed, 297 insertions(+), 18 deletions(-)
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 e422f3d8c6f..30c7a41e28d 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
@@ -94,15 +94,6 @@ public class OneWayReplicatorTest extends OneWayReplicatorTestBase {
super.cleanup();
}
- private void waitReplicatorStarted(String topicName) {
- Awaitility.await().untilAsserted(() -> {
- Optional<Topic> topicOptional2 = pulsar2.getBrokerService().getTopic(topicName, false).get();
- assertTrue(topicOptional2.isPresent());
- PersistentTopic persistentTopic2 = (PersistentTopic) topicOptional2.get();
- assertFalse(persistentTopic2.getProducers().isEmpty());
- });
- }
-
private void waitReplicatorStopped(String topicName) {
Awaitility.await().untilAsserted(() -> {
Optional<Topic> topicOptional2 = pulsar2.getBrokerService().getTopic(topicName, false).get();
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 b4eed00c447..317e43306e3 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
@@ -19,6 +19,8 @@
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 com.google.common.collect.Sets;
import java.net.URL;
import java.time.Duration;
@@ -29,6 +31,7 @@ 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.PulsarClient;
import org.apache.pulsar.common.naming.SystemTopicNames;
import org.apache.pulsar.common.policies.data.ClusterData;
@@ -55,7 +58,7 @@ public abstract class OneWayReplicatorTestBase extends TestRetrySupport {
protected ZookeeperServerTest brokerConfigZk1;
protected LocalBookkeeperEnsemble bkEnsemble1;
protected PulsarService pulsar1;
- protected BrokerService ns1;
+ protected BrokerService broker1;
protected PulsarAdmin admin1;
protected PulsarClient client1;
@@ -66,7 +69,7 @@ public abstract class OneWayReplicatorTestBase extends TestRetrySupport {
protected ZookeeperServerTest brokerConfigZk2;
protected LocalBookkeeperEnsemble bkEnsemble2;
protected PulsarService pulsar2;
- protected BrokerService ns2;
+ protected BrokerService broker2;
protected PulsarAdmin admin2;
protected PulsarClient client2;
@@ -89,23 +92,29 @@ public abstract class OneWayReplicatorTestBase extends TestRetrySupport {
setConfigDefaults(config1, cluster1, bkEnsemble1, brokerConfigZk1);
pulsar1 = new PulsarService(config1);
pulsar1.start();
- ns1 = pulsar1.getBrokerService();
-
+ broker1 = pulsar1.getBrokerService();
url1 = new URL(pulsar1.getWebServiceAddress());
urlTls1 = new URL(pulsar1.getWebServiceAddressTls());
- admin1 = PulsarAdmin.builder().serviceHttpUrl(url1.toString()).build();
- client1 = PulsarClient.builder().serviceUrl(url1.toString()).build();
// Start region 2
setConfigDefaults(config2, cluster2, bkEnsemble2, brokerConfigZk2);
pulsar2 = new PulsarService(config2);
pulsar2.start();
- ns2 = pulsar2.getBrokerService();
-
+ broker2 = pulsar2.getBrokerService();
url2 = new URL(pulsar2.getWebServiceAddress());
urlTls2 = new URL(pulsar2.getWebServiceAddressTls());
+ }
+
+ protected void startAdminClient() throws Exception {
+ admin1 = PulsarAdmin.builder().serviceHttpUrl(url1.toString()).build();
admin2 = PulsarAdmin.builder().serviceHttpUrl(url2.toString()).build();
- client2 = PulsarClient.builder().serviceUrl(url2.toString()).build();
+ }
+
+ protected void startPulsarClient() throws Exception{
+ ClientBuilder clientBuilder1 = PulsarClient.builder().serviceUrl(url1.toString());
+ client1 = initClient(clientBuilder1);
+ ClientBuilder clientBuilder2 = PulsarClient.builder().serviceUrl(url2.toString());
+ client2 = initClient(clientBuilder2);
}
protected void createDefaultTenantsAndClustersAndNamespace() throws Exception {
@@ -196,8 +205,12 @@ public abstract class OneWayReplicatorTestBase extends TestRetrySupport {
startBrokers();
+ startAdminClient();
+
createDefaultTenantsAndClustersAndNamespace();
+ startPulsarClient();
+
Thread.sleep(100);
log.info("--- OneWayReplicatorTestBase::setup completed ---");
}
@@ -287,4 +300,17 @@ public abstract class OneWayReplicatorTestBase extends TestRetrySupport {
config1 = new ServiceConfiguration();
config2 = new ServiceConfiguration();
}
+
+ protected void waitReplicatorStarted(String topicName) {
+ Awaitility.await().untilAsserted(() -> {
+ Optional<Topic> topicOptional2 = pulsar2.getBrokerService().getTopic(topicName, false).get();
+ assertTrue(topicOptional2.isPresent());
+ PersistentTopic persistentTopic2 = (PersistentTopic) topicOptional2.get();
+ assertFalse(persistentTopic2.getProducers().isEmpty());
+ });
+ }
+
+ protected PulsarClient initClient(ClientBuilder clientBuilder) throws Exception {
+ return clientBuilder.build();
+ }
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicationTxnTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicationTxnTest.java
new file mode 100644
index 00000000000..3caf4a1f239
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicationTxnTest.java
@@ -0,0 +1,262 @@
+/*
+ * 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 static org.apache.pulsar.common.naming.NamespaceName.SYSTEM_NAMESPACE;
+import static org.apache.pulsar.transaction.coordinator.impl.MLTransactionLogImpl.TRANSACTION_LOG_PREFIX;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertTrue;
+import com.google.common.collect.Sets;
+import java.util.Collections;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.broker.BrokerTestUtil;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.service.persistent.GeoPersistentReplicator;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.pulsar.broker.transaction.pendingack.impl.MLPendingAckStore;
+import org.apache.pulsar.client.api.ClientBuilder;
+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.Producer;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.transaction.Transaction;
+import org.apache.pulsar.common.api.proto.MessageMetadata;
+import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.naming.SystemTopicNames;
+import org.apache.pulsar.common.naming.TopicDomain;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
+import org.apache.pulsar.common.policies.data.TenantInfoImpl;
+import org.apache.pulsar.zookeeper.LocalBookkeeperEnsemble;
+import org.apache.pulsar.zookeeper.ZookeeperServerTest;
+import org.awaitility.reflect.WhiteboxImpl;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+@Slf4j
+@Test(groups = "broker")
+public class ReplicationTxnTest extends OneWayReplicatorTestBase {
+
+ private boolean transactionBufferSegmentedSnapshotEnabled = false;
+ private int txnLogPartitions = 4;
+
+ @Override
+ @BeforeClass(alwaysRun = true, timeOut = 300000)
+ public void setup() throws Exception {
+ super.setup();
+ }
+
+ @Override
+ @AfterClass(alwaysRun = true, timeOut = 300000)
+ public void cleanup() throws Exception {
+ super.cleanup();
+ }
+
+ @Override
+ protected PulsarClient initClient(ClientBuilder clientBuilder) throws Exception {
+ return clientBuilder.enableTransaction(true).build();
+ }
+
+ @Override
+ protected void setConfigDefaults(ServiceConfiguration config, String clusterName,
+ LocalBookkeeperEnsemble bookkeeperEnsemble, ZookeeperServerTest brokerConfigZk) {
+ super.setConfigDefaults(config, clusterName, bookkeeperEnsemble, brokerConfigZk);
+ config.setSystemTopicEnabled(true);
+ config.setTopicLevelPoliciesEnabled(true);
+ config.setTransactionCoordinatorEnabled(true);
+ config.setTransactionLogBatchedWriteEnabled(true);
+ config.setTransactionPendingAckBatchedWriteEnabled(true);
+ config.setTransactionBufferSegmentedSnapshotEnabled(transactionBufferSegmentedSnapshotEnabled);
+ }
+
+ @Override
+ protected void createDefaultTenantsAndClustersAndNamespace() throws Exception {
+ super.createDefaultTenantsAndClustersAndNamespace();
+
+ // Create resource that transaction function relies on.
+ admin1.tenants().createTenant(SYSTEM_NAMESPACE.getTenant(), new TenantInfoImpl(Collections.emptySet(),
+ Sets.newHashSet(cluster1, cluster2)));
+ admin1.namespaces().createNamespace(SYSTEM_NAMESPACE.toString(), 4);
+ pulsar1.getPulsarResources().getNamespaceResources().getPartitionedTopicResources().createPartitionedTopic(
+ SystemTopicNames.TRANSACTION_COORDINATOR_ASSIGN, new PartitionedTopicMetadata(txnLogPartitions));
+ //admin1.topics().createPartitionedTopic(SystemTopicNames.TRANSACTION_COORDINATOR_ASSIGN.toString(), 4);
+
+ admin2.tenants().createTenant(SYSTEM_NAMESPACE.getTenant(), new TenantInfoImpl(Collections.emptySet(),
+ Sets.newHashSet(cluster1, cluster2)));
+ admin2.namespaces().createNamespace(SYSTEM_NAMESPACE.toString(), 4);
+ pulsar2.getPulsarResources().getNamespaceResources().getPartitionedTopicResources().createPartitionedTopic(
+ SystemTopicNames.TRANSACTION_COORDINATOR_ASSIGN, new PartitionedTopicMetadata(txnLogPartitions));
+ }
+
+ private void pubAndSubOneMsg(String topic, String subscription) throws Exception {
+ Consumer consumer1 = client1.newConsumer(Schema.STRING).topic(topic).subscriptionName(subscription)
+ .isAckReceiptEnabled(true).subscribe();
+ Producer<String> producer1 = client1.newProducer(Schema.STRING).topic(topic).create();
+ producer1.newMessage().value("msg1").send();
+ // start txn.
+ Transaction txn = client1.newTransaction().withTransactionTimeout(1, TimeUnit.MINUTES).build().get();
+ // consume.
+ Message<String> c1Msg1 = consumer1.receive(5, TimeUnit.SECONDS);
+ assertNotNull(c1Msg1);
+ assertEquals(c1Msg1.getValue(), "msg1");
+ consumer1.acknowledgeAsync(c1Msg1.getMessageId(), txn).join();
+ // send.
+ producer1.newMessage(txn).value("msg2").send();
+ // commit.
+ txn.commit().get();
+
+ // Consume the msg with TXN.
+ Message<String> c1Msg2 = consumer1.receive(5, TimeUnit.SECONDS);
+ assertNotNull(c1Msg2);
+ assertEquals(c1Msg2.getValue(), "msg2");
+ consumer1.acknowledgeAsync(c1Msg2.getMessageId()).join();
+
+ // Consume messages on the remote cluster.
+ Consumer consumer2 = client2.newConsumer(Schema.STRING).topic(topic).subscriptionName(subscription).subscribe();
+ Message<String> c2Msg1 = consumer2.receive(15, TimeUnit.SECONDS);
+ assertNotNull(c2Msg1);
+ MessageMetadata msgMetadata1 = WhiteboxImpl.getInternalState(c2Msg1, "msgMetadata");
+ // Verify: the messages replicated has no TXN id.
+ assertFalse(msgMetadata1.hasTxnidMostBits());
+ assertFalse(msgMetadata1.hasTxnidLeastBits());
+ consumer2.acknowledge(c2Msg1);
+ Message<String> c2Msg2 = consumer2.receive(15, TimeUnit.SECONDS);
+ assertNotNull(c2Msg2);
+ MessageMetadata msgMetadata2 = WhiteboxImpl.getInternalState(c2Msg2, "msgMetadata");
+ // Verify: the messages replicated has no TXN id.
+ assertFalse(msgMetadata2.hasTxnidMostBits());
+ assertFalse(msgMetadata2.hasTxnidLeastBits());
+ consumer2.acknowledge(c2Msg2);
+
+ // cleanup.
+ producer1.close();
+ consumer1.close();
+ consumer2.close();
+ }
+
+ private void verifyNoReplicator(BrokerService broker, TopicName topicName) throws Exception {
+ String tpStr = topicName.toString();
+ CompletableFuture<Optional<Topic>> future = broker.getTopic(tpStr, true);
+ if (future == null) {
+ return;
+ }
+ PersistentTopic persistentTopic = (PersistentTopic) future.join().get();
+ assertTrue(persistentTopic.getReplicators().isEmpty());
+ }
+
+ @Test
+ public void testTxnLogNotBeReplicated() throws Exception {
+ final String topic = BrokerTestUtil.newUniqueName("persistent://" + replicatedNamespace + "/tp");
+ final String subscription = "s1";
+ admin1.topics().createNonPartitionedTopic(topic);
+ waitReplicatorStarted(topic);
+ admin1.topics().createSubscription(topic, subscription, MessageId.earliest);
+ admin2.topics().createSubscription(topic, subscription, MessageId.earliest);
+ // Pub & Sub.
+ pubAndSubOneMsg(topic, subscription);
+ // To cover more cases, sleep 3s.
+ Thread.sleep(3000);
+
+ // Verify: messages on the TXN system topic did not been replicated.
+ // __transaction_log_: it only uses ML, will not create topic.
+ for (int i = 0; i < txnLogPartitions; i++) {
+ TopicName txnLog = TopicName.get(TopicDomain.persistent.value(),
+ NamespaceName.SYSTEM_NAMESPACE, TRANSACTION_LOG_PREFIX + i);
+ assertNotNull(pulsar1.getManagedLedgerFactory()
+ .getManagedLedgerInfo(txnLog.getPersistenceNamingEncoding()));
+ assertFalse(broker1.getTopics().containsKey(txnLog.toString()));
+ }
+ // __transaction_pending_ack: it only uses ML, will not create topic.
+ TopicName pendingAck = TopicName.get(
+ MLPendingAckStore.getTransactionPendingAckStoreSuffix(topic, subscription));
+ assertNotNull(pulsar1.getManagedLedgerFactory()
+ .getManagedLedgerInfo(pendingAck.getPersistenceNamingEncoding()));
+ assertFalse(broker1.getTopics().containsKey(pendingAck.toString()));
+ // __transaction_buffer_snapshot.
+ verifyNoReplicator(broker1, TopicName.get(TopicDomain.persistent.value(),
+ TopicName.get(topic).getNamespaceObject(),
+ SystemTopicNames.TRANSACTION_BUFFER_SNAPSHOT));
+ verifyNoReplicator(broker1, TopicName.get(TopicDomain.persistent.value(),
+ TopicName.get(topic).getNamespaceObject(),
+ SystemTopicNames.TRANSACTION_BUFFER_SNAPSHOT_SEGMENTS));
+ verifyNoReplicator(broker1, TopicName.get(TopicDomain.persistent.value(),
+ TopicName.get(topic).getNamespaceObject(),
+ SystemTopicNames.TRANSACTION_BUFFER_SNAPSHOT_INDEXES));
+
+ // cleanup.
+ cleanupTopics(() -> {
+ admin1.topics().delete(topic);
+ admin2.topics().delete(topic);
+ try {
+ admin1.topics().delete(pendingAck.toString());
+ } catch (Exception ex) {}
+ try {
+ admin2.topics().delete(pendingAck.toString());
+ } catch (Exception ex) {}
+ });
+ }
+
+ @Test
+ public void testOngoingMessagesWillNotBeReplicated() throws Exception {
+ final String topic = BrokerTestUtil.newUniqueName("persistent://" + replicatedNamespace + "/tp");
+ final String subscription = "s1";
+ admin1.topics().createNonPartitionedTopic(topic);
+ waitReplicatorStarted(topic);
+ admin1.topics().createSubscription(topic, subscription, MessageId.earliest);
+ admin2.topics().createSubscription(topic, subscription, MessageId.earliest);
+ // Pub without commit.
+ Producer<String> producer1 = client1.newProducer(Schema.STRING).topic(topic).create();
+ Transaction txn = client1.newTransaction().withTransactionTimeout(1, TimeUnit.HOURS).build().get();
+ producer1.newMessage(txn).value("msg1").send();
+ // Verify: receive nothing on the remote cluster.
+ Consumer consumer2 = client2.newConsumer(Schema.STRING).topic(topic).subscriptionName(subscription).subscribe();
+ Message<String> msg = consumer2.receive(15, TimeUnit.SECONDS);
+ assertNull(msg);
+ // Verify: the repl cursor is not end of the topic.
+ PersistentTopic persistentTopic = (PersistentTopic) broker1.getTopic(topic, false).join().get();
+ GeoPersistentReplicator replicator =
+ (GeoPersistentReplicator) persistentTopic.getReplicators().values().iterator().next();
+ assertTrue(replicator.getCursor().hasMoreEntries());
+
+ // cleanup.
+ producer1.close();
+ consumer2.close();
+ cleanupTopics(() -> {
+ admin1.topics().delete(topic);
+ admin2.topics().delete(topic);
+ TopicName pendingAck = TopicName.get(
+ MLPendingAckStore.getTransactionPendingAckStoreSuffix(topic, subscription));
+ try {
+ admin1.topics().delete(pendingAck.toString());
+ } catch (Exception ex) {}
+ try {
+ admin2.topics().delete(pendingAck.toString());
+ } catch (Exception ex) {}
+ });
+ }
+}