From feb5e211f0f9202f0a4ab91b171f655d9c8d4cad Mon Sep 17 00:00:00 2001 From: coderzc Date: Mon, 10 Oct 2022 09:51:02 +0800 Subject: [PATCH] improve AdminApiTest --- .../pulsar/broker/admin/AdminApiTest.java | 131 +++++++++++++----- 1 file changed, 100 insertions(+), 31 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java index c75bb280faaa3..baaf7d15c0de6 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java @@ -139,8 +139,9 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.testng.Assert; +import org.testng.annotations.AfterClass; import org.testng.annotations.AfterMethod; -import org.testng.annotations.BeforeMethod; +import org.testng.annotations.BeforeClass; import org.testng.annotations.DataProvider; import org.testng.annotations.Test; @@ -163,7 +164,7 @@ public class AdminApiTest extends MockedPulsarServiceBaseTest { private NamespaceBundleFactory bundleFactory; - @BeforeMethod + @BeforeClass @Override public void setup() throws Exception { conf.setSystemTopicEnabled(false); @@ -192,14 +193,39 @@ public void setup() throws Exception { otherPulsar = mockPulsarSetup.getPulsar(); otheradmin = mockPulsarSetup.getAdmin(); - // Setup namespaces + setupClusters(); + } + + @AfterMethod(alwaysRun = true) + public void resetClusters() throws Exception { + pulsar.getConfiguration().setForceDeleteTenantAllowed(true); + pulsar.getConfiguration().setForceDeleteNamespaceAllowed(true); + for (String tenant : admin.tenants().getTenants()) { + for (String namespace : admin.namespaces().getNamespaces(tenant)) { + deleteNamespaceGraceFully(namespace, true); + } + admin.tenants().deleteTenant(tenant, true); + } + + for (String cluster : admin.clusters().getClusters()) { + admin.clusters().deleteCluster(cluster); + } + + pulsar.getConfiguration().setForceDeleteTenantAllowed(false); + pulsar.getConfiguration().setForceDeleteNamespaceAllowed(false); + + resetConfig(); + setupClusters(); + } + + private void setupClusters() throws PulsarAdminException { admin.clusters().createCluster("test", ClusterData.builder().serviceUrl(pulsar.getWebServiceAddress()).build()); TenantInfoImpl tenantInfo = new TenantInfoImpl(Set.of("role1", "role2"), Set.of("test")); admin.tenants().createTenant("prop-xyz", tenantInfo); admin.namespaces().createNamespace("prop-xyz/ns1", Set.of("test")); } - @AfterMethod(alwaysRun = true) + @AfterClass(alwaysRun = true) @Override public void cleanup() throws Exception { adminTls.close(); @@ -1190,6 +1216,7 @@ public void testGetStats() throws Exception { String subName = "my-sub"; // create consumer and subscription + @Cleanup Consumer consumer = pulsarClient.newConsumer().topic(topic).subscriptionName(subName).subscribe(); TopicStats topicStats = admin.topics().getStats(topic, false, false, true); @@ -1227,7 +1254,9 @@ public void testGetPartitionedStatsInternal() throws Exception { assertEquals(admin.topics().getPartitionedTopicMetadata(partitionedTopicName).partitions, 2); // create consumer and subscription - pulsarClient.newConsumer().topic(partitionedTopicName).subscriptionName(subName).subscribe(); + @Cleanup + Consumer consumer = + pulsarClient.newConsumer().topic(partitionedTopicName).subscriptionName(subName).subscribe(); // publish several messages publishMessagesOnPersistentTopic(partitionedTopicName, 10); @@ -1426,8 +1455,10 @@ public void testForceDeleteTenantNotAllowed() throws Exception { @Test public void testNamespaceSplitBundle() throws Exception { + admin.namespaces().createNamespace("prop-xyz/splitBundle", Set.of("test")); + // Force to create a topic - final String namespace = "prop-xyz/ns1"; + final String namespace = "prop-xyz/splitBundle"; final String topicName = (new StringBuilder("persistent://")).append(namespace).append("/ds2").toString(); Producer producer = pulsarClient.newProducer(Schema.BYTES) .topic(topicName) @@ -1489,7 +1520,9 @@ public void testNamespaceSplitBundleWithTopicCountEquallyDivideAlgorithm() throw for (int i = 0; i < bundles.getBundles().size(); i++) { assertNotEquals(bundles.getBundles().get(i).toString(), splitRange[i]); } - producers.forEach(Producer::closeAsync); + for (Producer producer : producers) { + producer.close(); + } } @Test @@ -1684,6 +1717,9 @@ public void testNamespaceSplitBundleWithInvalidAlgorithm() { @Test public void testNamespaceSplitBundleWithDefaultTopicCountEquallyDivideAlgorithm() throws Exception { + cleanup(); + setup(); + conf.setDefaultNamespaceBundleSplitAlgorithm(NamespaceBundleSplitAlgorithm.TOPIC_COUNT_EQUALLY_DIVIDE); // Force to create a topic final String namespace = "prop-xyz/ns1"; @@ -1717,7 +1753,9 @@ public void testNamespaceSplitBundleWithDefaultTopicCountEquallyDivideAlgorithm( for (int i = 0; i < bundles.getBundles().size(); i++) { assertNotEquals(bundles.getBundles().get(i).toString(), splitRange[i]); } - producers.forEach(Producer::closeAsync); + for (Producer producer : producers) { + producer.close(); + } conf.setDefaultNamespaceBundleSplitAlgorithm(NamespaceBundleSplitAlgorithm.RANGE_EQUALLY_DIVIDE_NAME); } @@ -1819,22 +1857,24 @@ public void testNamespaceSplitBundleConcurrent() throws Exception { @Test public void testNamespaceUnloadBundle() throws Exception { - assertEquals(admin.topics().getList("prop-xyz/ns1"), new ArrayList<>()); + admin.namespaces().createNamespace("prop-xyz/unloadBundle", Set.of("test")); + + assertEquals(admin.topics().getList("prop-xyz/unloadBundle"), new ArrayList<>()); // Force to create a topic - publishMessagesOnPersistentTopic("persistent://prop-xyz/ns1/ds2", 0); - assertEquals(admin.topics().getList("prop-xyz/ns1"), - Lists.newArrayList("persistent://prop-xyz/ns1/ds2")); + publishMessagesOnPersistentTopic("persistent://prop-xyz/unloadBundle/ds2", 0); + assertEquals(admin.topics().getList("prop-xyz/unloadBundle"), + Lists.newArrayList("persistent://prop-xyz/unloadBundle/ds2")); // create consumer and subscription - Consumer consumer = pulsarClient.newConsumer().topic("persistent://prop-xyz/ns1/ds2") + Consumer consumer = pulsarClient.newConsumer().topic("persistent://prop-xyz/unloadBundle/ds2") .subscriptionName("my-sub").subscribe(); - assertEquals(admin.topics().getSubscriptions("persistent://prop-xyz/ns1/ds2"), + assertEquals(admin.topics().getSubscriptions("persistent://prop-xyz/unloadBundle/ds2"), Lists.newArrayList("my-sub")); // Create producer Producer producer = pulsarClient.newProducer(Schema.BYTES) - .topic("persistent://prop-xyz/ns1/ds2") + .topic("persistent://prop-xyz/unloadBundle/ds2") .enableBatching(false) .messageRoutingMode(MessageRoutingMode.SinglePartition) .create(); @@ -1847,13 +1887,13 @@ public void testNamespaceUnloadBundle() throws Exception { producer.close(); try { - admin.namespaces().unloadNamespaceBundle("prop-xyz/ns1", "0x00000000_0xffffffff"); + admin.namespaces().unloadNamespaceBundle("prop-xyz/unloadBundle", "0x00000000_0xffffffff"); } catch (Exception e) { fail("Unload shouldn't have throw exception"); } // check that no one owns the namespace - NamespaceBundle bundle = bundleFactory.getBundle(NamespaceName.get("prop-xyz/ns1"), + NamespaceBundle bundle = bundleFactory.getBundle(NamespaceName.get("prop-xyz/unloadBundle"), Range.range(0L, BoundType.CLOSED, 0xffffffffL, BoundType.CLOSED)); assertFalse(pulsar.getNamespaceService().isServiceUnitOwned(bundle)); assertFalse(otherPulsar.getNamespaceService().isServiceUnitOwned(bundle)); @@ -1863,14 +1903,17 @@ public void testNamespaceUnloadBundle() throws Exception { // Force reload of namespace and wait for topic to be ready Awaitility.await().timeout(30, TimeUnit.SECONDS).ignoreExceptionsInstanceOf(PulsarAdminException.class) - .until(() -> admin.topics().getStats("persistent://prop-xyz/ns1/ds2") != null); + .until(() -> admin.topics().getStats("persistent://prop-xyz/unloadBundle/ds2") != null); - admin.topics().deleteSubscription("persistent://prop-xyz/ns1/ds2", "my-sub"); - admin.topics().delete("persistent://prop-xyz/ns1/ds2"); + admin.topics().deleteSubscription("persistent://prop-xyz/unloadBundle/ds2", "my-sub"); + admin.topics().delete("persistent://prop-xyz/unloadBundle/ds2"); } @Test(dataProvider = "numBundles") public void testNamespaceBundleUnload(Integer numBundles) throws Exception { + cleanup(); + setup(); + admin.namespaces().createNamespace("prop-xyz/ns1-bundles", numBundles); admin.namespaces().setNamespaceReplicationClusters("prop-xyz/ns1-bundles", Set.of("test")); @@ -1976,15 +2019,25 @@ public void testClearBacklogOnNamespace(Integer numBundles) throws Exception { admin.namespaces().setNamespaceReplicationClusters("prop-xyz/ns1-bundles", Set.of("test")); // create consumer and subscription - pulsarClient.newConsumer().topic("persistent://prop-xyz/ns1-bundles/ds2").subscriptionName("my-sub") - .subscribe(); - pulsarClient.newConsumer().topic("persistent://prop-xyz/ns1-bundles/ds2").subscriptionName("my-sub-1") - .subscribe(); - pulsarClient.newConsumer().topic("persistent://prop-xyz/ns1-bundles/ds2").subscriptionName("my-sub-2") + @Cleanup + Consumer consumer = + pulsarClient.newConsumer().topic("persistent://prop-xyz/ns1-bundles/ds2").subscriptionName("my-sub") + .subscribe(); + @Cleanup + Consumer consumer2 = + pulsarClient.newConsumer().topic("persistent://prop-xyz/ns1-bundles/ds2").subscriptionName("my-sub-1") + .subscribe(); + @Cleanup + Consumer consumer3 = + pulsarClient.newConsumer().topic("persistent://prop-xyz/ns1-bundles/ds2").subscriptionName("my-sub-2") .subscribe(); - pulsarClient.newConsumer().topic("persistent://prop-xyz/ns1-bundles/ds1").subscriptionName("my-sub") + @Cleanup + Consumer consumer4 = + pulsarClient.newConsumer().topic("persistent://prop-xyz/ns1-bundles/ds1").subscriptionName("my-sub") .subscribe(); - pulsarClient.newConsumer().topic("persistent://prop-xyz/ns1-bundles/ds1").subscriptionName("my-sub-1") + @Cleanup + Consumer consumer5 = + pulsarClient.newConsumer().topic("persistent://prop-xyz/ns1-bundles/ds1").subscriptionName("my-sub-1") .subscribe(); // Create producer @@ -2048,7 +2101,8 @@ public void testUnsubscribeOnNamespace(Integer numBundles) throws Exception { .subscriptionName("my-sub").subscribe(); Consumer consumer2 = pulsarClient.newConsumer().topic("persistent://prop-xyz/ns1-bundles/ds2") .subscriptionName("my-sub-1").subscribe(); - /* Consumer consumer3 = */ pulsarClient.newConsumer().topic("persistent://prop-xyz/ns1-bundles/ds2") + @Cleanup + Consumer consumer3 = pulsarClient.newConsumer().topic("persistent://prop-xyz/ns1-bundles/ds2") .subscriptionName("my-sub-2").subscribe(); Consumer consumer4 = pulsarClient.newConsumer().topic("persistent://prop-xyz/ns1-bundles/ds1") .subscriptionName("my-sub").subscribe(); @@ -2833,7 +2887,9 @@ public void testPulsarAdminForUriAndUrlEncoding(String topicName) throws Excepti final int numOfPartitions = 4; admin.topics().createPartitionedTopic(topic1, numOfPartitions); // Create a consumer to get stats on this topic - pulsarClient.newConsumer().topic(topic1).subscriptionName("my-subscriber-name").subscribe(); + @Cleanup + Consumer consumer = + pulsarClient.newConsumer().topic(topic1).subscriptionName("my-subscriber-name").subscribe(); TopicsImpl persistent = (TopicsImpl) admin.topics(); Field field = TopicsImpl.class.getDeclaredField("adminV2Topics"); @@ -3089,12 +3145,14 @@ public void testCompactionStatus() throws Exception { public void testTopicStatsLastExpireTimestampForSubscription() throws PulsarAdminException, PulsarClientException, InterruptedException { admin.namespaces().setNamespaceMessageTTL("prop-xyz/ns1", 10); final String topic = "persistent://prop-xyz/ns1/testTopicStatsLastExpireTimestampForSubscription"; + @Cleanup Producer producer = pulsarClient.newProducer() .topic(topic) .create(); for (int i = 0; i < 10; i++) { producer.send(new byte[1024 * i * 5]); } + @Cleanup Consumer consumer = pulsarClient.newConsumer() .topic(topic) .subscriptionName("sub-1") @@ -3169,10 +3227,12 @@ public void testCreateAndDeleteNamespaceWithBundles() throws Exception { @Test public void testBacklogSizeShouldBeZeroWhenConsumerAckedAllMessages() throws Exception { final String topic = "persistent://prop-xyz/ns1/testBacklogSizeShouldBeZeroWhenConsumerAckedAllMessages"; + @Cleanup Consumer consumer = pulsarClient.newConsumer() .topic(topic) .subscriptionName("sub-1") .subscribe(); + @Cleanup Producer producer = pulsarClient.newProducer() .topic(topic) .create(); @@ -3208,6 +3268,7 @@ public void testGetTtlDurationDefaultInSeconds() throws Exception { public void testGetReadPositionWhenJoining() throws Exception { final String topic = "persistent://prop-xyz/ns1/testGetReadPositionWhenJoining-" + UUID.randomUUID().toString(); final String subName = "my-sub"; + @Cleanup Producer producer = pulsarClient.newProducer() .topic(topic) .enableBatching(false) @@ -3219,12 +3280,14 @@ public void testGetReadPositionWhenJoining() throws Exception { messageId = (MessageIdImpl) producer.send(("Hello Pulsar - " + i).getBytes()); } + List> consumers = new ArrayList<>(); for (int i = 0; i < 2; i++) { - pulsarClient.newConsumer() + Consumer consumer = pulsarClient.newConsumer() .topic(topic) .subscriptionType(SubscriptionType.Key_Shared) .subscriptionName(subName) .subscribe(); + consumers.add(consumer); } TopicStats stats = admin.topics().getStats(topic); @@ -3235,6 +3298,10 @@ public void testGetReadPositionWhenJoining() throws Exception { ConsumerStats consumerStats = subStats.getConsumers().get(0); Assert.assertEquals(consumerStats.getReadPositionWhenJoining(), PositionImpl.get(messageId.getLedgerId(), messageId.getEntryId() + 1).toString()); + + for (Consumer consumer : consumers) { + consumer.close(); + } } @Test @@ -3247,13 +3314,15 @@ public void testPartitionedTopicMsgDelayedAggregated() throws Exception { admin.topics().createPartitionedTopic(topic, numPartitions); for (int i = 0; i < 2; i++) { - pulsarClient.newConsumer() + @Cleanup + Consumer consumer = pulsarClient.newConsumer() .topic(topic) .subscriptionType(SubscriptionType.Shared) .subscriptionName(subName) .subscribe(); } + @Cleanup Producer producer = pulsarClient.newProducer() .topic(topic) .enableBatching(false)