You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by zi...@apache.org on 2025/04/24 06:24:44 UTC
(pulsar) branch master updated: [cleanup][test] Cleanup OneWayReplicatorUsingGlobalPartitionedTest and OneWayReplicatorUsingGlobalZKTest (#24200)
This is an automated email from the ASF dual-hosted git repository.
zixuan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 5965c29f78e [cleanup][test] Cleanup OneWayReplicatorUsingGlobalPartitionedTest and OneWayReplicatorUsingGlobalZKTest (#24200)
5965c29f78e is described below
commit 5965c29f78e036850b435f87bb8750b070318bcd
Author: Zixuan Liu <no...@gmail.com>
AuthorDate: Thu Apr 24 14:24:38 2025 +0800
[cleanup][test] Cleanup OneWayReplicatorUsingGlobalPartitionedTest and OneWayReplicatorUsingGlobalZKTest (#24200)
Signed-off-by: Zixuan Liu <no...@gmail.com>
---
.../broker/service/OneWayReplicatorTest.java | 2 +-
...OneWayReplicatorUsingGlobalPartitionedTest.java | 209 +++++++++++----------
.../service/OneWayReplicatorUsingGlobalZKTest.java | 186 ++++++++++++------
3 files changed, 232 insertions(+), 165 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 d3356a60695..8a24195642e 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
@@ -105,7 +105,7 @@ import org.testng.annotations.Test;
@Slf4j
@Test(groups = "broker")
-public class OneWayReplicatorTest extends OneWayReplicatorTestBase {
+public final class OneWayReplicatorTest extends OneWayReplicatorTestBase {
@Override
@BeforeClass(alwaysRun = true, timeOut = 300000)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalPartitionedTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalPartitionedTest.java
index 2a2a1befd16..0c875313b18 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalPartitionedTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalPartitionedTest.java
@@ -18,31 +18,44 @@
*/
package org.apache.pulsar.broker.service;
+import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.fail;
+import com.github.benmanes.caffeine.cache.AsyncLoadingCache;
+import java.lang.reflect.Field;
import java.time.Duration;
import java.util.Arrays;
import java.util.HashSet;
import java.util.Map;
import java.util.Optional;
+import java.util.Set;
+import java.util.UUID;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
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.resources.ClusterResources;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.policies.data.ClusterData;
+import org.apache.pulsar.common.policies.data.TenantInfo;
import org.apache.pulsar.common.policies.data.TopicType;
import org.apache.pulsar.zookeeper.LocalBookkeeperEnsemble;
import org.apache.pulsar.zookeeper.ZookeeperServerTest;
import org.awaitility.Awaitility;
+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 OneWayReplicatorUsingGlobalPartitionedTest extends OneWayReplicatorTest {
+public class OneWayReplicatorUsingGlobalPartitionedTest extends OneWayReplicatorTestBase {
@Override
@BeforeClass(alwaysRun = true, timeOut = 300000)
@@ -65,108 +78,6 @@ public class OneWayReplicatorUsingGlobalPartitionedTest extends OneWayReplicator
config.setDefaultNumPartitions(1);
}
- @Override
- @Test(enabled = false)
- public void testReplicatorProducerStatInTopic() throws Exception {
- super.testReplicatorProducerStatInTopic();
- }
-
- @Override
- @Test(enabled = false)
- public void testCreateRemoteConsumerFirst() throws Exception {
- super.testReplicatorProducerStatInTopic();
- }
-
- @Override
- @Test(enabled = false)
- public void testTopicCloseWhenInternalProducerCloseErrorOnce() throws Exception {
- super.testReplicatorProducerStatInTopic();
- }
-
- @Override
- @Test(enabled = false)
- public void testConcurrencyOfUnloadBundleAndRecreateProducer() throws Exception {
- super.testConcurrencyOfUnloadBundleAndRecreateProducer();
- }
-
- @Override
- @Test(enabled = false)
- public void testPartitionedTopicLevelReplication() throws Exception {
- super.testPartitionedTopicLevelReplication();
- }
-
- @Override
- @Test(enabled = false)
- public void testPartitionedTopicLevelReplicationRemoteTopicExist() throws Exception {
- super.testPartitionedTopicLevelReplicationRemoteTopicExist();
- }
-
- @Override
- @Test(enabled = false)
- public void testPartitionedTopicLevelReplicationRemoteConflictTopicExist() throws Exception {
- super.testPartitionedTopicLevelReplicationRemoteConflictTopicExist();
- }
-
- @Override
- @Test(enabled = false)
- public void testConcurrencyOfUnloadBundleAndRecreateProducer2() throws Exception {
- super.testConcurrencyOfUnloadBundleAndRecreateProducer2();
- }
-
- @Override
- @Test(enabled = false)
- public void testUnFenceTopicToReuse() throws Exception {
- super.testUnFenceTopicToReuse();
- }
-
- @Override
- @Test(enabled = false)
- public void testDeleteNonPartitionedTopic() throws Exception {
- super.testDeleteNonPartitionedTopic();
- }
-
- @Override
- @Test(enabled = false)
- public void testDeletePartitionedTopic() throws Exception {
- super.testDeletePartitionedTopic();
- }
-
- @Override
- @Test(enabled = false)
- public void testNoExpandTopicPartitionsWhenDisableTopicLevelReplication() throws Exception {
- super.testNoExpandTopicPartitionsWhenDisableTopicLevelReplication();
- }
-
- @Override
- @Test(enabled = false)
- public void testExpandTopicPartitionsOnNamespaceLevelReplication() throws Exception {
- super.testExpandTopicPartitionsOnNamespaceLevelReplication();
- }
-
- @Override
- @Test(enabled = false)
- public void testReloadWithTopicLevelGeoReplication(ReplicationLevel replicationLevel) throws Exception {
- super.testReloadWithTopicLevelGeoReplication(replicationLevel);
- }
-
- @Test(enabled = false)
- @Override
- public void testConfigReplicationStartAt() throws Exception {
- super.testConfigReplicationStartAt();
- }
-
- @Test(enabled = false)
- @Override
- public void testDifferentTopicCreationRule(ReplicationMode replicationMode) throws Exception {
- super.testDifferentTopicCreationRule(replicationMode);
- }
-
- @Test(enabled = false)
- @Override
- public void testReplicationCountMetrics() throws Exception {
- super.testReplicationCountMetrics();
- }
-
@Test(timeOut = 60_000)
public void testRemoveCluster() throws Exception {
// Initialize.
@@ -204,4 +115,96 @@ public class OneWayReplicatorUsingGlobalPartitionedTest extends OneWayReplicator
admin2.topics().delete(topic);
admin2.namespaces().deleteNamespace(ns1);
}
+
+ /**
+ * This test used to confirm the "start replicator retry task" will be skipped after the topic is closed.
+ */
+ @Test
+ public void testCloseTopicAfterStartReplicationFailed() throws Exception {
+ Field fieldTopicNameCache = TopicName.class.getDeclaredField("cache");
+ fieldTopicNameCache.setAccessible(true);
+ ConcurrentHashMap<String, TopicName> topicNameCache =
+ (ConcurrentHashMap<String, TopicName>) fieldTopicNameCache.get(null);
+ final String topicName = BrokerTestUtil.newUniqueName("persistent://" + nonReplicatedNamespace + "/tp_");
+ // 1.Create topic, does not enable replication now.
+ admin1.topics().createNonPartitionedTopic(topicName);
+ Producer<byte[]> producer1 = client1.newProducer().topic(topicName).create();
+ PersistentTopic persistentTopic =
+ (PersistentTopic) pulsar1.getBrokerService().getTopic(topicName, false).join().get();
+
+ // We inject an error to make "start replicator" to fail.
+ AsyncLoadingCache<String, Boolean> existsCache =
+ WhiteboxImpl.getInternalState(pulsar1.getConfigurationMetadataStore(), "existsCache");
+ String path = "/admin/partitioned-topics/" + TopicName.get(topicName).getPersistenceNamingEncoding();
+ existsCache.put(path, CompletableFuture.completedFuture(true));
+
+ // 2.Enable replication and unload topic after failed to start replicator.
+ admin1.topics().setReplicationClusters(topicName, Arrays.asList(cluster1, cluster2));
+ Thread.sleep(3000);
+ producer1.close();
+ existsCache.synchronous().invalidate(path);
+ admin1.topics().unload(topicName);
+ // Verify: the "start replicator retry task" will be skipped after the topic is closed.
+ // - Retry delay is "PersistentTopic.POLICY_UPDATE_FAILURE_RETRY_TIME_SECONDS": 60s, so wait for 70s.
+ // - Since the topic should not be touched anymore, we use "TopicName" to confirm whether it be used by
+ // Replication again.
+ Thread.sleep(10 * 1000);
+ topicNameCache.remove(topicName);
+ Thread.sleep(60 * 1000);
+ assertTrue(!topicNameCache.containsKey(topicName));
+
+ // cleanup.
+ admin1.topics().setReplicationClusters(topicName, Arrays.asList(cluster1));
+ admin1.topics().delete(topicName, false);
+ }
+
+ // https://github.com/apache/pulsar/issues/22967
+ @Test
+ public void testPartitionedTopicWithTopicPolicyAndNoReplicationClusters() throws Exception {
+ final String topicName = BrokerTestUtil.newUniqueName("persistent://" + replicatedNamespace + "/tp_");
+ admin1.topics().createPartitionedTopic(topicName, 2);
+ try {
+ admin1.topicPolicies().setMessageTTL(topicName, 5);
+ Awaitility.await().ignoreExceptions().untilAsserted(() -> {
+ assertEquals(admin2.topics().getPartitionedTopicMetadata(topicName).partitions, 2);
+ });
+ admin1.topics().updatePartitionedTopic(topicName, 3, false);
+ Awaitility.await().ignoreExceptions().untilAsserted(() -> {
+ assertEquals(admin2.topics().getPartitionedTopicMetadata(topicName).partitions, 3);
+ });
+ } finally {
+ // cleanup.
+ admin1.topics().deletePartitionedTopic(topicName, true);
+ }
+ }
+
+ @Test(timeOut = 30 * 1000)
+ public void testCreateRemoteAdminFailed() throws Exception {
+ final TenantInfo tenantInfo = admin1.tenants().getTenantInfo(defaultTenant);
+ final String ns1 = defaultTenant + "/ns_" + UUID.randomUUID().toString().replace("-", "");
+ final String randomClusterName = "c_" + UUID.randomUUID().toString().replace("-", "");
+ final String topic = BrokerTestUtil.newUniqueName(ns1 + "/tp");
+ admin1.namespaces().createNamespace(ns1);
+ admin1.topics().createPartitionedTopic(topic, 2);
+
+ // Inject a wrong cluster data which with empty fields.
+ ClusterResources clusterResources = broker1.getPulsar().getPulsarResources().getClusterResources();
+ clusterResources.createCluster(randomClusterName, ClusterData.builder().build());
+ Set<String> allowedClusters = new HashSet<>(tenantInfo.getAllowedClusters());
+ allowedClusters.add(randomClusterName);
+ admin1.tenants().updateTenant(defaultTenant, TenantInfo.builder().adminRoles(tenantInfo.getAdminRoles())
+ .allowedClusters(allowedClusters).build());
+
+ // Verify.
+ try {
+ admin1.topics().setReplicationClusters(topic, Arrays.asList(cluster1, randomClusterName));
+ fail("Expected a error due to empty fields");
+ } catch (Exception ex) {
+ // Expected an error.
+ }
+
+ // cleanup.
+ admin1.topics().deletePartitionedTopic(topic);
+ admin1.tenants().updateTenant(defaultTenant, tenantInfo);
+ }
}
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 a530baab9be..1be8faa9669 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
@@ -22,30 +22,40 @@ import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.fail;
+import com.github.benmanes.caffeine.cache.AsyncLoadingCache;
+import java.lang.reflect.Field;
import java.time.Duration;
import java.util.Arrays;
import java.util.HashSet;
import java.util.Map;
import java.util.Optional;
+import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.broker.BrokerTestUtil;
+import org.apache.pulsar.broker.resources.ClusterResources;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
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.Schema;
import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.policies.data.ClusterData;
import org.apache.pulsar.common.policies.data.RetentionPolicies;
+import org.apache.pulsar.common.policies.data.TenantInfo;
import org.awaitility.Awaitility;
+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 OneWayReplicatorUsingGlobalZKTest extends OneWayReplicatorTest {
+public class OneWayReplicatorUsingGlobalZKTest extends OneWayReplicatorTestBase {
@Override
@BeforeClass(alwaysRun = true, timeOut = 300000)
@@ -60,78 +70,144 @@ public class OneWayReplicatorUsingGlobalZKTest extends OneWayReplicatorTest {
super.cleanup();
}
- @Test(enabled = false)
- public void testReplicatorProducerStatInTopic() throws Exception {
- super.testReplicatorProducerStatInTopic();
- }
+ /**
+ * This test used to confirm the "start replicator retry task" will be skipped after the topic is closed.
+ */
+ @Test
+ public void testCloseTopicAfterStartReplicationFailed() throws Exception {
+ Field fieldTopicNameCache = TopicName.class.getDeclaredField("cache");
+ fieldTopicNameCache.setAccessible(true);
+ ConcurrentHashMap<String, TopicName> topicNameCache =
+ (ConcurrentHashMap<String, TopicName>) fieldTopicNameCache.get(null);
+ final String topicName = BrokerTestUtil.newUniqueName("persistent://" + nonReplicatedNamespace + "/tp_");
+ // 1.Create topic, does not enable replication now.
+ admin1.topics().createNonPartitionedTopic(topicName);
+ Producer<byte[]> producer1 = client1.newProducer().topic(topicName).create();
+ PersistentTopic persistentTopic =
+ (PersistentTopic) pulsar1.getBrokerService().getTopic(topicName, false).join().get();
- @Test(enabled = false)
- public void testCreateRemoteConsumerFirst() throws Exception {
- super.testReplicatorProducerStatInTopic();
- }
+ // We inject an error to make "start replicator" to fail.
+ AsyncLoadingCache<String, Boolean> existsCache =
+ WhiteboxImpl.getInternalState(pulsar1.getConfigurationMetadataStore(), "existsCache");
+ String path = "/admin/partitioned-topics/" + TopicName.get(topicName).getPersistenceNamingEncoding();
+ existsCache.put(path, CompletableFuture.completedFuture(true));
- @Test(enabled = false)
- public void testTopicCloseWhenInternalProducerCloseErrorOnce() throws Exception {
- super.testReplicatorProducerStatInTopic();
- }
+ // 2.Enable replication and unload topic after failed to start replicator.
+ admin1.topics().setReplicationClusters(topicName, Arrays.asList(cluster1, cluster2));
+ Thread.sleep(3000);
+ producer1.close();
+ existsCache.synchronous().invalidate(path);
+ admin1.topics().unload(topicName);
+ // Verify: the "start replicator retry task" will be skipped after the topic is closed.
+ // - Retry delay is "PersistentTopic.POLICY_UPDATE_FAILURE_RETRY_TIME_SECONDS": 60s, so wait for 70s.
+ // - Since the topic should not be touched anymore, we use "TopicName" to confirm whether it be used by
+ // Replication again.
+ Thread.sleep(10 * 1000);
+ topicNameCache.remove(topicName);
+ Thread.sleep(60 * 1000);
+ assertTrue(!topicNameCache.containsKey(topicName));
- @Test(enabled = false)
- public void testConcurrencyOfUnloadBundleAndRecreateProducer() throws Exception {
- super.testConcurrencyOfUnloadBundleAndRecreateProducer();
+ // cleanup.
+ admin1.topics().setReplicationClusters(topicName, Arrays.asList(cluster1));
+ admin1.topics().delete(topicName, false);
}
- @Test(enabled = false)
- public void testPartitionedTopicLevelReplication() throws Exception {
- super.testPartitionedTopicLevelReplication();
+ // https://github.com/apache/pulsar/issues/22967
+ @Test
+ public void testPartitionedTopicWithTopicPolicyAndNoReplicationClusters() throws Exception {
+ final String topicName = BrokerTestUtil.newUniqueName("persistent://" + replicatedNamespace + "/tp_");
+ admin1.topics().createPartitionedTopic(topicName, 2);
+ try {
+ admin1.topicPolicies().setMessageTTL(topicName, 5);
+ Awaitility.await().ignoreExceptions().untilAsserted(() -> {
+ assertEquals(admin2.topics().getPartitionedTopicMetadata(topicName).partitions, 2);
+ });
+ admin1.topics().updatePartitionedTopic(topicName, 3, false);
+ Awaitility.await().ignoreExceptions().untilAsserted(() -> {
+ assertEquals(admin2.topics().getPartitionedTopicMetadata(topicName).partitions, 3);
+ });
+ } finally {
+ // cleanup.
+ admin1.topics().deletePartitionedTopic(topicName, true);
+ }
}
- @Test(enabled = false)
- public void testPartitionedTopicLevelReplicationRemoteTopicExist() throws Exception {
- super.testPartitionedTopicLevelReplicationRemoteTopicExist();
- }
+ @Test(timeOut = 30 * 1000)
+ public void testCreateRemoteAdminFailed() throws Exception {
+ final TenantInfo tenantInfo = admin1.tenants().getTenantInfo(defaultTenant);
+ final String ns1 = defaultTenant + "/ns_" + UUID.randomUUID().toString().replace("-", "");
+ final String randomClusterName = "c_" + UUID.randomUUID().toString().replace("-", "");
+ final String topic = BrokerTestUtil.newUniqueName(ns1 + "/tp");
+ admin1.namespaces().createNamespace(ns1);
+ admin1.topics().createPartitionedTopic(topic, 2);
- @Test(enabled = false)
- public void testPartitionedTopicLevelReplicationRemoteConflictTopicExist() throws Exception {
- super.testPartitionedTopicLevelReplicationRemoteConflictTopicExist();
- }
+ // Inject a wrong cluster data which with empty fields.
+ ClusterResources clusterResources = broker1.getPulsar().getPulsarResources().getClusterResources();
+ clusterResources.createCluster(randomClusterName, ClusterData.builder().build());
+ Set<String> allowedClusters = new HashSet<>(tenantInfo.getAllowedClusters());
+ allowedClusters.add(randomClusterName);
+ admin1.tenants().updateTenant(defaultTenant, TenantInfo.builder().adminRoles(tenantInfo.getAdminRoles())
+ .allowedClusters(allowedClusters).build());
- @Test(enabled = false)
- public void testConcurrencyOfUnloadBundleAndRecreateProducer2() throws Exception {
- super.testConcurrencyOfUnloadBundleAndRecreateProducer2();
- }
+ // Verify.
+ try {
+ admin1.topics().setReplicationClusters(topic, Arrays.asList(cluster1, randomClusterName));
+ fail("Expected a error due to empty fields");
+ } catch (Exception ex) {
+ // Expected an error.
+ }
- @Test(enabled = false)
- public void testUnFenceTopicToReuse() throws Exception {
- super.testUnFenceTopicToReuse();
+ // cleanup.
+ admin1.topics().deletePartitionedTopic(topic);
+ admin1.tenants().updateTenant(defaultTenant, tenantInfo);
}
@Test
public void testDeleteNonPartitionedTopic() throws Exception {
- super.testDeleteNonPartitionedTopic();
+ 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 {
- super.testDeletePartitionedTopic();
- }
+ final String topicName = BrokerTestUtil.newUniqueName("persistent://" + replicatedNamespace + "/tp_");
+ admin1.topics().createPartitionedTopic(topicName, 2);
- @Test(enabled = false)
- public void testNoExpandTopicPartitionsWhenDisableTopicLevelReplication() throws Exception {
- super.testNoExpandTopicPartitionsWhenDisableTopicLevelReplication();
- }
+ // Verify replicator works.
+ verifyReplicationWorks(topicName);
- @Test(enabled = false)
- public void testExpandTopicPartitionsOnNamespaceLevelReplication() throws Exception {
- super.testExpandTopicPartitionsOnNamespaceLevelReplication();
- }
+ // Disable replication.
+ setTopicLevelClusters(topicName, Arrays.asList(cluster1), admin1, pulsar1);
+ setTopicLevelClusters(topicName, Arrays.asList(cluster2), admin2, pulsar2);
+
+ // Delete topic.
+ admin1.topics().deletePartitionedTopic(topicName);
- @Test(enabled = false)
- public void testReloadWithTopicLevelGeoReplication(ReplicationLevel replicationLevel) throws Exception {
- super.testReloadWithTopicLevelGeoReplication(replicationLevel);
+ // Verify the topic was deleted.
+ assertFalse(pulsar1.getPulsarResources().getNamespaceResources().getPartitionedTopicResources()
+ .partitionedTopicExists(TopicName.get(topicName)));
+ assertFalse(pulsar2.getPulsarResources().getNamespaceResources().getPartitionedTopicResources()
+ .partitionedTopicExists(TopicName.get(topicName)));
}
@Test
- @Override
public void testConfigReplicationStartAt() throws Exception {
// Initialize.
String ns1 = defaultTenant + "/ns_" + UUID.randomUUID().toString().replace("-", "");
@@ -174,18 +250,6 @@ public class OneWayReplicatorUsingGlobalZKTest extends OneWayReplicatorTest {
});
}
- @Test(enabled = false)
- @Override
- public void testDifferentTopicCreationRule(ReplicationMode replicationMode) throws Exception {
- super.testDifferentTopicCreationRule(replicationMode);
- }
-
- @Test(enabled = false)
- @Override
- public void testReplicationCountMetrics() throws Exception {
- super.testReplicationCountMetrics();
- }
-
@Test
public void testRemoveCluster() throws Exception {
// Initialize.