From de786f632798eddc1ec4c8c65c622884ae39d9a4 Mon Sep 17 00:00:00 2001 From: Cong Zhao Date: Thu, 13 Oct 2022 12:20:43 +0800 Subject: [PATCH 1/9] [improve][test] Improve AdminApiTest to reduce the execution time (#17980) (cherry picked from commit e4d3c494677b04aee7196c431c3d50835c5044f8) --- .../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 7206de3609072..3f151daf8d207 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 @@ -135,8 +135,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; @@ -159,7 +160,7 @@ public class AdminApiTest extends MockedPulsarServiceBaseTest { private NamespaceBundleFactory bundleFactory; - @BeforeMethod + @BeforeClass @Override public void setup() throws Exception { conf.setLoadBalancerEnabled(true); @@ -186,14 +187,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(Sets.newHashSet("role1", "role2"), Sets.newHashSet("test")); admin.tenants().createTenant("prop-xyz", tenantInfo); admin.namespaces().createNamespace("prop-xyz/ns1", Sets.newHashSet("test")); } - @AfterMethod(alwaysRun = true) + @AfterClass(alwaysRun = true) @Override public void cleanup() throws Exception { adminTls.close(); @@ -1167,6 +1193,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); @@ -1204,7 +1231,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); @@ -1403,8 +1432,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) @@ -1466,7 +1497,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 @@ -1483,6 +1516,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"; @@ -1516,7 +1552,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); } @@ -1618,22 +1656,24 @@ public void testNamespaceSplitBundleConcurrent() throws Exception { @Test public void testNamespaceUnloadBundle() throws Exception { - assertEquals(admin.topics().getList("prop-xyz/ns1"), Lists.newArrayList()); + 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(); @@ -1646,13 +1686,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)); @@ -1662,14 +1702,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", Sets.newHashSet("test")); @@ -1775,15 +1818,25 @@ public void testClearBacklogOnNamespace(Integer numBundles) throws Exception { admin.namespaces().setNamespaceReplicationClusters("prop-xyz/ns1-bundles", Sets.newHashSet("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 @@ -1847,7 +1900,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(); @@ -2606,7 +2660,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"); @@ -2861,12 +2917,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") @@ -2941,10 +2999,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(); @@ -2980,6 +3040,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) @@ -2991,12 +3052,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); @@ -3007,6 +3070,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 @@ -3019,13 +3086,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) From a3e593abbcb3a88509816ccf584823478f21aa74 Mon Sep 17 00:00:00 2001 From: fengyubiao <9947090@qq.com> Date: Thu, 24 Nov 2022 15:20:44 +0800 Subject: [PATCH 2/9] [fix][test] Fix flaky test deleteNamespaceGracefully (#18220) (cherry picked from commit c544ea3aae1a8be8e7bcd4f89fb1899ce0a3350f) --- .../pulsar/broker/admin/AdminApi2Test.java | 8 +- .../pulsar/broker/admin/AdminApiTest.java | 38 ++--- .../broker/admin/AdminApiTlsAuthTest.java | 2 +- .../auth/MockedPulsarServiceBaseTest.java | 142 ++++++++++++++++++ .../NamespaceOwnershipListenerTests.java | 13 +- .../OwnerShipForCurrentServerTestBase.java | 3 +- .../ResourceGroupConfigListenerTest.java | 2 +- .../service/BacklogQuotaManagerTest.java | 25 ++- ...erServiceBundlesCacheInvalidationTest.java | 2 +- .../service/PersistentTopicE2ETest.java | 4 +- .../transaction/TransactionTestBase.java | 21 ++- .../api/DispatcherBlockConsumerTest.java | 3 +- 12 files changed, 221 insertions(+), 42 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java index fcc2f14490784..3c99452fdc07d 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java @@ -1386,7 +1386,7 @@ public void testDeleteTenant() throws Exception { assertTrue(admin.topics().getList(namespace).isEmpty()); // delete namespace - admin.namespaces().deleteNamespace(namespace, false); + deleteNamespaceWithRetry(namespace, false); assertFalse(admin.namespaces().getNamespaces(tenant).contains(namespace)); assertTrue(admin.namespaces().getNamespaces(tenant).isEmpty()); @@ -1475,7 +1475,7 @@ public void testDeleteNamespace(NamespaceAttr namespaceAttr) throws Exception { admin.topics().deletePartitionedTopic(topic); // delete namespace - admin.namespaces().deleteNamespace(namespace, false); + deleteNamespaceWithRetry(namespace, false); assertFalse(admin.namespaces().getNamespaces(tenant).contains(namespace)); assertTrue(admin.namespaces().getNamespaces(tenant).isEmpty()); @@ -1572,7 +1572,7 @@ public void testDeleteNamespaceWithTopicPolicies() throws Exception { }); producer.close(); admin.topics().delete(topic); - admin.namespaces().deleteNamespace(namespace); + deleteNamespaceWithRetry(namespace, false); Awaitility.await().untilAsserted(() -> { assertTrue(admin.namespaces().getNamespaces(tenant).isEmpty()); }); @@ -1783,7 +1783,7 @@ public void testForceDeleteNamespace() throws Exception { final String topic = "persistent://" + namespaceName + "/test" + UUID.randomUUID(); pulsarClient.newProducer(Schema.DOUBLE).topic(topic).create().close(); Awaitility.await().untilAsserted(() -> assertNotNull(admin.schemas().getSchemaInfo(topic))); - admin.namespaces().deleteNamespace(namespaceName, true); + deleteNamespaceWithRetry(namespaceName, true); try { admin.schemas().getSchemaInfo(topic); } catch (PulsarAdminException e) { 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 3f151daf8d207..a9ba361bbb9ca 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 @@ -196,7 +196,7 @@ public void resetClusters() throws Exception { pulsar.getConfiguration().setForceDeleteNamespaceAllowed(true); for (String tenant : admin.tenants().getTenants()) { for (String namespace : admin.namespaces().getNamespaces(tenant)) { - deleteNamespaceGraceFully(namespace, true); + deleteNamespaceWithRetry(namespace, true); } admin.tenants().deleteTenant(tenant, true); } @@ -291,7 +291,7 @@ public void clusters() throws Exception { Awaitility.await() .untilAsserted(() -> assertEquals(admin.clusters().getClusters(), Lists.newArrayList("test"))); - admin.namespaces().deleteNamespace("prop-xyz/ns1"); + deleteNamespaceWithRetry("prop-xyz/ns1", false); admin.clusters().deleteCluster("test"); assertEquals(admin.clusters().getClusters(), Lists.newArrayList()); @@ -516,7 +516,7 @@ public void brokers() throws Exception { String.format("%s:%d", parts[0], pulsar.getListenPortHTTPS().get())); Assert.assertEquals(nsMap2.size(), 2); - admin.namespaces().deleteNamespace("prop-xyz/ns1"); + deleteNamespaceWithRetry("prop-xyz/ns1", false); admin.clusters().deleteCluster("test"); assertEquals(admin.clusters().getClusters(), Lists.newArrayList()); } @@ -710,7 +710,7 @@ public void testGetDynamicLocalConfiguration() throws Exception { } @Test - public void properties() throws PulsarAdminException { + public void properties() throws Exception { try { admin.tenants().getTenantInfo("does-not-exist"); fail("should have failed"); @@ -739,7 +739,7 @@ public void properties() throws PulsarAdminException { assertEquals(e.getStatusCode(), 409); assertEquals(e.getMessage(), "The tenant still has active namespaces"); } - admin.namespaces().deleteNamespace("prop-xyz/ns1"); + deleteNamespaceWithRetry("prop-xyz/ns1", false); admin.tenants().deleteTenant("prop-xyz"); assertEquals(admin.tenants().getTenants(), Lists.newArrayList()); @@ -768,7 +768,7 @@ public void namespaces() throws Exception { assertEquals(admin.namespaces().getPolicies("prop-xyz/ns3").bundles.getNumBundles(), 4); assertEquals(admin.namespaces().getPolicies("prop-xyz/ns3").bundles.getBoundaries().size(), 5); - admin.namespaces().deleteNamespace("prop-xyz/ns3"); + deleteNamespaceWithRetry("prop-xyz/ns3", false); try { admin.namespaces().createNamespace("non-existing/ns1"); @@ -842,7 +842,7 @@ public void namespaces() throws Exception { } assertTrue(i < 10); - admin.namespaces().deleteNamespace("prop-xyz/ns1"); + deleteNamespaceWithRetry("prop-xyz/ns1", false); assertEquals(admin.namespaces().getNamespaces("prop-xyz"), Lists.newArrayList("prop-xyz/ns2")); try { @@ -1266,7 +1266,7 @@ public void testGetPartitionedStatsInternal() throws Exception { @Test(dataProvider = "numBundles") public void testDeleteNamespaceBundle(Integer numBundles) throws Exception { - admin.namespaces().deleteNamespace("prop-xyz/ns1"); + deleteNamespaceWithRetry("prop-xyz/ns1", false); admin.namespaces().createNamespace("prop-xyz/ns1-bundles", numBundles); admin.namespaces().setNamespaceReplicationClusters("prop-xyz/ns1-bundles", Sets.newHashSet("test")); @@ -1278,8 +1278,8 @@ public void testDeleteNamespaceBundle(Integer numBundles) throws Exception { assertEquals(admin.namespaces().getTopics("prop-xyz/ns1-bundles"), Lists.newArrayList()); - admin.namespaces().deleteNamespace("prop-xyz/ns1-bundles"); - assertEquals(admin.namespaces().getNamespaces("prop-xyz", "test"), Lists.newArrayList()); + deleteNamespaceWithRetry("prop-xyz/ns1-bundles", false); + assertEquals(admin.namespaces().getNamespaces("prop-xyz", "test"), new ArrayList<>()); } @Test @@ -1372,7 +1372,7 @@ public void testDeleteNamespaceForcefully() throws Exception { } // delete namespace forcefully - admin.namespaces().deleteNamespace(namespace, true); + deleteNamespaceWithRetry(namespace, true); assertFalse(admin.namespaces().getNamespaces(tenant).contains(namespace)); assertTrue(admin.namespaces().getNamespaces(tenant).isEmpty()); @@ -1432,7 +1432,7 @@ public void testForceDeleteTenantNotAllowed() throws Exception { @Test public void testNamespaceSplitBundle() throws Exception { - admin.namespaces().createNamespace("prop-xyz/splitBundle", Set.of("test")); + admin.namespaces().createNamespace("prop-xyz/splitBundle", Sets.newHashSet("test")); // Force to create a topic final String namespace = "prop-xyz/splitBundle"; @@ -1656,7 +1656,7 @@ public void testNamespaceSplitBundleConcurrent() throws Exception { @Test public void testNamespaceUnloadBundle() throws Exception { - admin.namespaces().createNamespace("prop-xyz/unloadBundle", Set.of("test")); + admin.namespaces().createNamespace("prop-xyz/unloadBundle", Sets.newHashSet("test")); assertEquals(admin.topics().getList("prop-xyz/unloadBundle"), new ArrayList<>()); @@ -2086,7 +2086,7 @@ public void testBackwardCompatibility() throws Exception { assertEquals(result.someNewIntField, 0); assertNull(result.someNewString); - admin.namespaces().deleteNamespace("prop-xyz/ns1"); + deleteNamespaceWithRetry("prop-xyz/ns1", false); admin.tenants().deleteTenant("prop-xyz"); assertEquals(admin.tenants().getTenants(), Lists.newArrayList()); } @@ -2974,9 +2974,9 @@ public void testSubscriptionExpiry() throws Exception { admin.topics().delete(topic1); admin.topics().delete(topic2); admin.topics().delete(topic3); - admin.namespaces().deleteNamespace(namespace1); - admin.namespaces().deleteNamespace(namespace2); - admin.namespaces().deleteNamespace(namespace3); + deleteNamespaceWithRetry(namespace1, false); + deleteNamespaceWithRetry(namespace2, false); + deleteNamespaceWithRetry(namespace3, false); } @Test @@ -2989,11 +2989,11 @@ public void testCreateAndDeleteNamespaceWithBundles() throws Exception { String ns = BrokerTestUtil.newUniqueName("prop-xyz/ns"); admin.namespaces().createNamespace(ns, 24); - admin.namespaces().deleteNamespace(ns); + deleteNamespaceWithRetry(ns, false); // Re-create and re-delete admin.namespaces().createNamespace(ns, 32); - admin.namespaces().deleteNamespace(ns); + deleteNamespaceWithRetry(ns, false); } @Test diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTlsAuthTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTlsAuthTest.java index 311e31be735da..11b806a656fdb 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTlsAuthTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTlsAuthTest.java @@ -443,7 +443,7 @@ public void testDeleteNamespace() throws Exception { admin.topics().delete("tenant1/ns1/foobar", true); log.info("Deleting namespace"); - admin.namespaces().deleteNamespace("tenant1/ns1"); + deleteNamespaceWithRetry("tenant1/ns1", false, admin); } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java index 84964680d3472..4f3af27a38449 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java @@ -33,13 +33,20 @@ import java.net.URL; import java.nio.charset.StandardCharsets; import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Date; import java.util.List; import java.util.Map; import java.util.Optional; import java.util.Set; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import java.util.function.Consumer; import java.util.function.Predicate; import java.util.function.Supplier; +import javax.ws.rs.container.AsyncResponse; +import javax.ws.rs.container.TimeoutHandler; import org.apache.bookkeeper.client.BookKeeper; import org.apache.bookkeeper.client.EnsemblePlacementPolicy; import org.apache.bookkeeper.client.PulsarMockBookKeeper; @@ -51,6 +58,7 @@ import org.apache.pulsar.broker.ServiceConfiguration; import org.apache.pulsar.broker.intercept.CounterBrokerInterceptor; import org.apache.pulsar.broker.namespace.NamespaceService; +import org.apache.pulsar.broker.service.BrokerTestBase; import org.apache.pulsar.client.admin.PulsarAdmin; import org.apache.pulsar.client.admin.PulsarAdminBuilder; import org.apache.pulsar.client.admin.PulsarAdminException; @@ -67,6 +75,7 @@ import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.MockZooKeeper; import org.apache.zookeeper.data.ACL; +import org.awaitility.Awaitility; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.testng.annotations.DataProvider; @@ -498,6 +507,139 @@ protected void setupDefaultTenantAndNamespace() throws Exception { } } + protected Object asyncRequests(Consumer function) throws Exception { + TestAsyncResponse ctx = new TestAsyncResponse(); + function.accept(ctx); + ctx.latch.await(); + if (ctx.e != null) { + throw (Exception) ctx.e; + } + return ctx.response; + } + + public static class TestAsyncResponse implements AsyncResponse { + + Object response; + Throwable e; + CountDownLatch latch = new CountDownLatch(1); + + @Override + public boolean resume(Object response) { + this.response = response; + latch.countDown(); + return true; + } + + @Override + public boolean resume(Throwable response) { + this.e = response; + latch.countDown(); + return true; + } + + @Override + public boolean cancel() { + return false; + } + + @Override + public boolean cancel(int retryAfter) { + return false; + } + + @Override + public boolean cancel(Date retryAfter) { + return false; + } + + @Override + public boolean isSuspended() { + return false; + } + + @Override + public boolean isCancelled() { + return false; + } + + @Override + public boolean isDone() { + return false; + } + + @Override + public boolean setTimeout(long time, TimeUnit unit) { + return false; + } + + @Override + public void setTimeoutHandler(TimeoutHandler handler) { + + } + + @Override + public Collection> register(Class callback) { + return null; + } + + @Override + public Map, Collection>> register(Class callback, Class... callbacks) { + return null; + } + + @Override + public Collection> register(Object callback) { + return null; + } + + @Override + public Map, Collection>> register(Object callback, Object... callbacks) { + return null; + } + + } + + /** + * see {@link BrokerTestBase#deleteNamespaceWithRetry(String, boolean, PulsarAdmin, Collection)} + */ + protected void deleteNamespaceWithRetry(String ns, boolean force) + throws Exception { + BrokerTestBase.deleteNamespaceWithRetry(ns, force, admin, pulsar); + } + + /** + * see {@link BrokerTestBase#deleteNamespaceWithRetry(String, boolean, PulsarAdmin, Collection)} + */ + protected void deleteNamespaceWithRetry(String ns, boolean force, PulsarAdmin admin) + throws Exception { + BrokerTestBase.deleteNamespaceWithRetry(ns, force, admin, pulsar); + } + + /** + * see {@link MockedPulsarServiceBaseTest#deleteNamespaceWithRetry(String, boolean, PulsarAdmin, Collection)} + */ + public static void deleteNamespaceWithRetry(String ns, boolean force, PulsarAdmin admin, PulsarService...pulsars) + throws Exception { + deleteNamespaceWithRetry(ns, force, admin, Arrays.asList(pulsars)); + } + + /** + * 1. Pause system "__change_event" topic creates. + * 2. Do delete namespace with retry because maybe fail by race-condition with create topics. + */ + public static void deleteNamespaceWithRetry(String ns, boolean force, PulsarAdmin admin, + Collection pulsars) throws Exception { + Awaitility.await().atMost(5, TimeUnit.SECONDS).until(() -> { + try { + // Maybe fail by race-condition with create topics, just retry. + admin.namespaces().deleteNamespace(ns, force); + return true; + } catch (Exception ex) { + return false; + } + }); + } + @DataProvider(name = "invalidPersistentPolicies") public Object[][] incorrectPersistentPolicies() { return new Object[][] { diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceOwnershipListenerTests.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceOwnershipListenerTests.java index c0a468816281d..2738ab897d2a6 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceOwnershipListenerTests.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceOwnershipListenerTests.java @@ -55,7 +55,7 @@ protected void cleanup() throws Exception { } @Test - public void testNamespaceBundleOwnershipListener() throws PulsarAdminException, InterruptedException, PulsarClientException { + public void testNamespaceBundleOwnershipListener() throws Exception { final CountDownLatch countDownLatch = new CountDownLatch(2); final AtomicBoolean onLoad = new AtomicBoolean(false); @@ -101,11 +101,11 @@ public void unLoad(NamespaceBundle bundle) { Assert.assertTrue(onLoad.get()); Assert.assertTrue(unLoad.get()); admin.topics().delete(topic); - admin.namespaces().deleteNamespace(namespace); + deleteNamespaceWithRetry(namespace, false); } @Test - public void testGetAllPartitions() throws PulsarAdminException, ExecutionException, InterruptedException { + public void testGetAllPartitions() throws Exception { final String namespace = "prop/" + UUID.randomUUID().toString(); admin.namespaces().createNamespace(namespace, Sets.newHashSet("test")); assertTrue(admin.namespaces().getNamespaces("prop").contains(namespace)); @@ -122,12 +122,11 @@ public void testGetAllPartitions() throws PulsarAdminException, ExecutionExcepti } admin.topics().deletePartitionedTopic(topicName); - admin.namespaces().deleteNamespace(namespace); + deleteNamespaceWithRetry(namespace, false); } @Test - public void testNamespaceBundleLookupOnwershipListener() throws PulsarAdminException, InterruptedException, - PulsarClientException { + public void testNamespaceBundleLookupOnwershipListener() throws Exception { final CountDownLatch countDownLatch = new CountDownLatch(2); final AtomicInteger onLoad = new AtomicInteger(0); final AtomicInteger unLoad = new AtomicInteger(0); @@ -172,6 +171,6 @@ public boolean test(NamespaceBundle namespaceBundle) { Assert.assertEquals(onLoad.get(), 1); Assert.assertEquals(unLoad.get(), 1); admin.topics().delete(topic); - admin.namespaces().deleteNamespace(namespace); + deleteNamespaceWithRetry(namespace, false); } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/OwnerShipForCurrentServerTestBase.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/OwnerShipForCurrentServerTestBase.java index c25c6c23fb011..80558361f4be8 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/OwnerShipForCurrentServerTestBase.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/OwnerShipForCurrentServerTestBase.java @@ -130,7 +130,8 @@ protected void setupBrokerMocks(PulsarService pulsar) throws Exception { MockZooKeeperSession mockZooKeeperSession = MockZooKeeperSession.newInstance(mockZooKeeper); doReturn(new ZKMetadataStore(mockZooKeeperSession)).when(pulsar).createLocalMetadataStore(); doReturn(new ZKMetadataStore(mockZooKeeperSession)).when(pulsar).createConfigurationMetadataStore(); - Supplier namespaceServiceSupplier = () -> spyWithClassAndConstructorArgs(NamespaceService.class, pulsar); + Supplier namespaceServiceSupplier = () -> spyWithClassAndConstructorArgs( + NamespaceService.class, pulsar); doReturn(namespaceServiceSupplier).when(pulsar).getNamespaceServiceProvider(); SameThreadOrderedSafeExecutor executor = new SameThreadOrderedSafeExecutor(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupConfigListenerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupConfigListenerTest.java index 38cae3253fe33..64c5fee14d057 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupConfigListenerTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupConfigListenerTest.java @@ -197,7 +197,7 @@ public void testResourceGroupAttachToNamespace() throws Exception { assertNull(pulsar.getResourceGroupServiceManager() .getNamespaceResourceGroup(NamespaceName.get(namespaceName)))); - admin.namespaces().deleteNamespace(namespaceName); + deleteNamespaceWithRetry(namespaceName, false); deleteResourceGroup(rgName); } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BacklogQuotaManagerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BacklogQuotaManagerTest.java index 1d2878ca64471..96fc452745bf7 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BacklogQuotaManagerTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BacklogQuotaManagerTest.java @@ -40,6 +40,7 @@ import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl; import org.apache.pulsar.broker.PulsarService; import org.apache.pulsar.broker.ServiceConfiguration; +import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest; import org.apache.pulsar.broker.service.persistent.PersistentTopic; import org.apache.pulsar.client.admin.GetStatsOptions; import org.apache.pulsar.client.admin.PulsarAdmin; @@ -82,6 +83,22 @@ public class BacklogQuotaManagerTest { private static final int TIME_TO_CHECK_BACKLOG_QUOTA = 2; private static final int MAX_ENTRIES_PER_LEDGER = 5; + /** + * see {@link MockedPulsarServiceBaseTest#deleteNamespaceWithRetry(String, boolean, PulsarAdmin, Collection)} + */ + protected void deleteNamespaceWithRetry(String ns, boolean force) + throws Exception { + MockedPulsarServiceBaseTest.deleteNamespaceWithRetry(ns, force, admin, pulsar); + } + + /** + * see {@link MockedPulsarServiceBaseTest#deleteNamespaceWithRetry(String, boolean, PulsarAdmin, Collection)} + */ + protected void deleteNamespaceWithRetry(String ns, boolean force, PulsarAdmin admin) + throws Exception { + MockedPulsarServiceBaseTest.deleteNamespaceWithRetry(ns, force, admin, pulsar); + } + @DataProvider(name = "backlogQuotaSizeGB") public Object[][] backlogQuotaSizeGB() { return new Object[][] { { true }, { false } }; @@ -158,10 +175,10 @@ void createNamespaces() throws PulsarAdminException { } @AfterMethod(alwaysRun = true) - void clearNamespaces() throws PulsarAdminException { - admin.namespaces().deleteNamespace("prop/ns-quota", true); - admin.namespaces().deleteNamespace("prop/quotahold", true); - admin.namespaces().deleteNamespace("prop/quotaholdasync", true); + void clearNamespaces() throws Exception { + deleteNamespaceWithRetry("prop/ns-quota", true); + deleteNamespaceWithRetry("prop/quotahold", true); + deleteNamespaceWithRetry("prop/quotaholdasync", true); } private void rolloverStats() { diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceBundlesCacheInvalidationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceBundlesCacheInvalidationTest.java index 573d473ed587a..a55540c3f4f9d 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceBundlesCacheInvalidationTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceBundlesCacheInvalidationTest.java @@ -59,7 +59,7 @@ public void testRecreateNamespace() throws Exception { // Delete and recreate with 32 bundles admin.topics().delete(topic); - admin.namespaces().deleteNamespace(namespace, false); + deleteNamespaceWithRetry(namespace, false); admin.namespaces().createNamespace(namespace, 32); BundlesData bundlesData = admin.namespaces().getBundles(namespace); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicE2ETest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicE2ETest.java index 54dc706016bbf..fdd1170c21b38 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicE2ETest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicE2ETest.java @@ -958,7 +958,7 @@ public void testMessageExpiry() throws Exception { consumer.close(); admin.topics().deleteSubscription(topicName, subName); admin.topics().delete(topicName); - admin.namespaces().deleteNamespace(namespaceName); + deleteNamespaceWithRetry(namespaceName, false); } @Test @@ -1052,7 +1052,7 @@ public void testMessageExpiryWithTopicMessageTTL() throws Exception { consumer.close(); admin.topics().deleteSubscription(topicName, subName); admin.topics().delete(topicName); - admin.namespaces().deleteNamespace(namespaceName, true); + deleteNamespaceWithRetry(namespaceName, true); } catch (PulsarAdminException e) { Assert.assertEquals(e.getStatusCode(), 500); } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTestBase.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTestBase.java index 25c555f09b949..990f63011b320 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTestBase.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTestBase.java @@ -44,9 +44,11 @@ import org.apache.pulsar.broker.BookKeeperClientFactory; import org.apache.pulsar.broker.PulsarService; import org.apache.pulsar.broker.ServiceConfiguration; +import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest; import org.apache.pulsar.broker.auth.SameThreadOrderedSafeExecutor; import org.apache.pulsar.broker.intercept.CounterBrokerInterceptor; import org.apache.pulsar.broker.namespace.NamespaceService; +import org.apache.pulsar.broker.service.BrokerTestBase; import org.apache.pulsar.client.admin.PulsarAdmin; import org.apache.pulsar.client.admin.PulsarAdminBuilder; import org.apache.pulsar.client.api.ClientBuilder; @@ -198,7 +200,8 @@ protected void setupBrokerMocks(PulsarService pulsar) throws Exception { MockZooKeeperSession mockZooKeeperSession = MockZooKeeperSession.newInstance(mockZooKeeper); doReturn(new ZKMetadataStore(mockZooKeeperSession)).when(pulsar).createLocalMetadataStore(); doReturn(new ZKMetadataStore(mockZooKeeperSession)).when(pulsar).createConfigurationMetadataStore(); - Supplier namespaceServiceSupplier = () -> spyWithClassAndConstructorArgs(NamespaceService.class, pulsar); + Supplier namespaceServiceSupplier = + () -> spyWithClassAndConstructorArgs(NamespaceService.class, pulsar); doReturn(namespaceServiceSupplier).when(pulsar).getNamespaceServiceProvider(); SameThreadOrderedSafeExecutor executor = new SameThreadOrderedSafeExecutor(); @@ -330,4 +333,20 @@ protected final void internalCleanup() { log.warn("Failed to clean up mocked pulsar service:", e); } } + + /** + * see {@link BrokerTestBase#deleteNamespaceWithRetry(String, boolean, PulsarAdmin, Collection)} + */ + protected void deleteNamespaceWithRetry(String ns, boolean force) + throws Exception { + MockedPulsarServiceBaseTest.deleteNamespaceWithRetry(ns, force, admin, pulsarServiceList); + } + + /** + * see {@link MockedPulsarServiceBaseTest#deleteNamespaceWithRetry(String, boolean, PulsarAdmin, Collection)} + */ + protected void deleteNamespaceWithRetry(String ns, boolean force, PulsarAdmin admin) + throws Exception { + MockedPulsarServiceBaseTest.deleteNamespaceWithRetry(ns, force, admin, pulsarServiceList); + } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DispatcherBlockConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DispatcherBlockConsumerTest.java index 1a56d689799be..df695cbf715a8 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DispatcherBlockConsumerTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DispatcherBlockConsumerTest.java @@ -631,7 +631,8 @@ public void testBrokerSubscriptionRecovery(boolean unloadBundleGracefully) throw // if broker unload bundle gracefully then cursor metadata recovered from zk else from ledger if (unloadBundleGracefully) { // set clean namespace which will not let broker unload bundle gracefully: stop broker - Supplier namespaceServiceSupplier = () -> spyWithClassAndConstructorArgs(NamespaceService.class, pulsar); + Supplier namespaceServiceSupplier = + () -> spyWithClassAndConstructorArgs(NamespaceService.class, pulsar); doReturn(namespaceServiceSupplier).when(pulsar).getNamespaceServiceProvider(); } stopBroker(); From 69b8955f4375198fc784b506e00827f3cb6b51de Mon Sep 17 00:00:00 2001 From: xiangying <1984997880@qq.com> Date: Sat, 25 Feb 2023 18:13:01 +0800 Subject: [PATCH 3/9] rollback some redundant file and resolve checkstyle --- .../pulsar/broker/web/PulsarWebResource.java | 2 +- .../NamespaceOwnershipListenerTests.java | 3 - .../elastic/ElasticSearchJavaRestClient.java | 224 ------------------ .../io/sinks/ElasticSearch7SinkTester.java | 41 ---- .../io/sinks/ElasticSearch8SinkTester.java | 43 ---- .../io/sinks/OpenSearchSinkTester.java | 88 ------- 6 files changed, 1 insertion(+), 400 deletions(-) delete mode 100644 pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/client/elastic/ElasticSearchJavaRestClient.java delete mode 100644 tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/ElasticSearch7SinkTester.java delete mode 100644 tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/ElasticSearch8SinkTester.java delete mode 100644 tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/OpenSearchSinkTester.java diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java index 68e5b7da8243b..976ba5d328e75 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java @@ -21,10 +21,10 @@ import static com.google.common.base.Preconditions.checkArgument; import static java.util.concurrent.TimeUnit.SECONDS; import static org.apache.commons.lang3.StringUtils.isBlank; +import com.fasterxml.jackson.databind.ObjectMapper; import com.github.benmanes.caffeine.cache.CacheLoader; import com.github.benmanes.caffeine.cache.Caffeine; import com.github.benmanes.caffeine.cache.LoadingCache; -import com.fasterxml.jackson.databind.ObjectMapper; import com.google.common.collect.BoundType; import com.google.common.collect.Lists; import com.google.common.collect.Range; diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceOwnershipListenerTests.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceOwnershipListenerTests.java index 2738ab897d2a6..bf0e252660a71 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceOwnershipListenerTests.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceOwnershipListenerTests.java @@ -21,9 +21,7 @@ import com.google.common.collect.Sets; import java.util.concurrent.atomic.AtomicInteger; import org.apache.pulsar.broker.service.BrokerTestBase; -import org.apache.pulsar.client.admin.PulsarAdminException; import org.apache.pulsar.client.api.Producer; -import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.common.naming.NamespaceBundle; import org.apache.pulsar.common.naming.NamespaceName; import org.testng.Assert; @@ -34,7 +32,6 @@ import java.util.List; import java.util.UUID; import java.util.concurrent.CountDownLatch; -import java.util.concurrent.ExecutionException; import java.util.concurrent.atomic.AtomicBoolean; import static org.testng.Assert.assertTrue; diff --git a/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/client/elastic/ElasticSearchJavaRestClient.java b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/client/elastic/ElasticSearchJavaRestClient.java deleted file mode 100644 index e420d7a5cef99..0000000000000 --- a/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/client/elastic/ElasticSearchJavaRestClient.java +++ /dev/null @@ -1,224 +0,0 @@ -/* - * 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.io.elasticsearch.client.elastic; - -import co.elastic.clients.elasticsearch.ElasticsearchClient; -import co.elastic.clients.elasticsearch._types.ElasticsearchException; -import co.elastic.clients.elasticsearch._types.Result; -import co.elastic.clients.elasticsearch.core.DeleteRequest; -import co.elastic.clients.elasticsearch.core.DeleteResponse; -import co.elastic.clients.elasticsearch.core.IndexRequest; -import co.elastic.clients.elasticsearch.core.IndexResponse; -import co.elastic.clients.elasticsearch.core.SearchRequest; -import co.elastic.clients.elasticsearch.core.SearchResponse; -import co.elastic.clients.elasticsearch.indices.CreateIndexRequest; -import co.elastic.clients.elasticsearch.indices.CreateIndexResponse; -import co.elastic.clients.elasticsearch.indices.DeleteIndexRequest; -import co.elastic.clients.elasticsearch.indices.ExistsRequest; -import co.elastic.clients.elasticsearch.indices.IndexSettings; -import co.elastic.clients.elasticsearch.indices.RefreshRequest; -import co.elastic.clients.json.jackson.JacksonJsonpMapper; -import co.elastic.clients.transport.ElasticsearchTransport; -import co.elastic.clients.transport.rest_client.RestClientTransport; -import com.fasterxml.jackson.annotation.JsonInclude; -import com.fasterxml.jackson.databind.ObjectMapper; -import com.fasterxml.jackson.databind.SerializationFeature; -import com.google.common.annotations.VisibleForTesting; -import java.io.IOException; -import java.util.Map; -import java.util.Objects; -import lombok.extern.slf4j.Slf4j; -import org.apache.http.HttpHost; -import org.apache.pulsar.io.elasticsearch.ElasticSearchConfig; -import org.apache.pulsar.io.elasticsearch.client.BulkProcessor; -import org.apache.pulsar.io.elasticsearch.client.RestClient; -import org.elasticsearch.client.Node; -import org.elasticsearch.client.RestClientBuilder; -import org.opensearch.action.bulk.BulkProcessor; - -@Slf4j -public class ElasticSearchJavaRestClient extends RestClient { - - private final ElasticsearchClient client; - private final ObjectMapper objectMapper = new ObjectMapper() - .configure(SerializationFeature.INDENT_OUTPUT, false) - .setSerializationInclusion(JsonInclude.Include.ALWAYS); - private BulkProcessor bulkProcessor; - private ElasticsearchTransport transport; - - @VisibleForTesting - public void setBulkProcessor(BulkProcessor bulkProcessor) { - this.bulkProcessor = bulkProcessor; - } - - @VisibleForTesting - public void setTransport(ElasticsearchTransport transport) { - this.transport = transport; - } - - public ElasticSearchJavaRestClient(ElasticSearchConfig elasticSearchConfig, - BulkProcessor.Listener bulkProcessorListener) { - super(elasticSearchConfig, bulkProcessorListener); - - log.info("ElasticSearch URL {}", config.getElasticSearchUrl()); - final HttpHost[] httpHosts = getHttpHosts(); - - RestClientBuilder builder = org.elasticsearch.client.RestClient.builder(httpHosts) - .setRequestConfigCallback(builder1 -> builder1 - .setContentCompressionEnabled(config.isCompressionEnabled()) - .setConnectionRequestTimeout(config.getConnectionRequestTimeoutInMs()) - .setConnectTimeout(config.getConnectTimeoutInMs()) - .setSocketTimeout(config.getSocketTimeoutInMs())) - .setHttpClientConfigCallback(this.configCallback) - .setFailureListener(new org.elasticsearch.client.RestClient.FailureListener() { - public void onFailure(Node node) { - log.warn("Node host={} failed", node.getHost()); - } - }); - transport = new RestClientTransport(builder.build(), new JacksonJsonpMapper(objectMapper)); - client = new ElasticsearchClient(transport); - if (elasticSearchConfig.isBulkEnabled()) { - bulkProcessor = new ElasticBulkProcessor(elasticSearchConfig, client, bulkProcessorListener); - } else { - bulkProcessor = null; - } - } - - @Override - public boolean indexExists(String index) throws IOException { - final ExistsRequest request = new ExistsRequest.Builder() - .index(index) - .build(); - return client.indices().exists(request).value(); - } - - @Override - public boolean createIndex(String index) throws IOException { - final CreateIndexRequest createIndexRequest = new CreateIndexRequest.Builder() - .index(index) - .settings(new IndexSettings.Builder() - .numberOfShards(config.getIndexNumberOfShards() + "") - .numberOfReplicas(config.getIndexNumberOfReplicas() + "") - .build() - ) - .build(); - try { - final CreateIndexResponse createIndexResponse = client.indices().create(createIndexRequest); - if ((createIndexResponse.acknowledged()) - && createIndexResponse.shardsAcknowledged()) { - return true; - } - throw new IOException("Unable to create index, acknowledged: " + createIndexResponse.acknowledged() - + " shardsAcknowledged: " + createIndexResponse.shardsAcknowledged()); - } catch (ElasticsearchException ex) { - final String errorType = Objects.requireNonNull(ex.response().error().type()); - if (errorType.contains("resource_already_exists_exception")) { - return false; - } - throw ex; - } - } - - @Override - public boolean deleteIndex(String index) throws IOException { - return client.indices().delete(new DeleteIndexRequest.Builder().index(index).build()).acknowledged(); - } - - @Override - public boolean deleteDocument(String index, String documentId) throws IOException { - final DeleteRequest req = new - DeleteRequest.Builder() - .index(config.getIndexName()) - .id(documentId) - .build(); - - DeleteResponse deleteResponse = client.delete(req); - return deleteResponse.result().equals(Result.Deleted) || deleteResponse.result().equals(Result.NotFound); - } - - @Override - public boolean indexDocument(String index, String documentId, String documentSource) throws IOException { - final Map mapped = objectMapper.readValue(documentSource, Map.class); - final IndexRequest indexRequest = new IndexRequest.Builder<>() - .index(config.getIndexName()) - .document(mapped) - .id(documentId) - .build(); - final IndexResponse indexResponse = client.index(indexRequest); - - return indexResponse.result().equals(Result.Created) || indexResponse.result().equals(Result.Updated); - } - - public SearchResponse search(String indexName) throws IOException { - return search(indexName, "*:*"); - } - - @VisibleForTesting - public SearchResponse search(String indexName, String query) throws IOException { - final RefreshRequest refreshRequest = new RefreshRequest.Builder().index(indexName).build(); - client.indices().refresh(refreshRequest); - - query = query.replace("/", "\\/"); - return client.search(new SearchRequest.Builder().index(indexName) - .q(query) - .build(), Map.class); - } - - @Override - public long totalHits(String indexName) throws IOException { - return totalHits(indexName, "*:*"); - } - - @Override - public long totalHits(String indexName, String query) throws IOException { - final SearchResponse searchResponse = search(indexName, query); - return searchResponse.hits().total().value(); - } - - @Override - public BulkProcessor getBulkProcessor() { - if (bulkProcessor == null) { - throw new IllegalStateException("bulkProcessor not enabled"); - } - return bulkProcessor; - } - - @Override - public void closeClient() { - if (bulkProcessor != null) { - bulkProcessor.close(); - } - // client doesn't need to be closed, only the transport instance - try { - transport.close(); - } catch (IOException e) { - log.warn("error while closing the client", e); - } - } - - @VisibleForTesting - public ElasticsearchClient getClient() { - return client; - } - - @VisibleForTesting - public ElasticsearchTransport getTransport() { - return transport; - } -} diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/ElasticSearch7SinkTester.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/ElasticSearch7SinkTester.java deleted file mode 100644 index 65b38c677bfc5..0000000000000 --- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/ElasticSearch7SinkTester.java +++ /dev/null @@ -1,41 +0,0 @@ -/* - * 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.tests.integration.io.sinks; - -import java.util.Optional; -import org.apache.pulsar.tests.integration.topologies.PulsarCluster; -import org.testcontainers.elasticsearch.ElasticsearchContainer; - -public class ElasticSearch7SinkTester extends ElasticSearchSinkTester { - - public static final String ELASTICSEARCH_7 = Optional.ofNullable(System.getenv("ELASTICSEARCH_IMAGE_V7")) - .orElse("docker.elastic.co/elasticsearch/elasticsearch:7.17.7"); - - - public ElasticSearch7SinkTester(boolean schemaEnable) { - super(schemaEnable); - } - - @Override - protected ElasticsearchContainer createSinkService(PulsarCluster cluster) { - return new ElasticsearchContainer(ELASTICSEARCH_7) - .withEnv("ES_JAVA_OPTS", "-Xms128m -Xmx256m"); - } - -} diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/ElasticSearch8SinkTester.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/ElasticSearch8SinkTester.java deleted file mode 100644 index bb52c4ff03fea..0000000000000 --- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/ElasticSearch8SinkTester.java +++ /dev/null @@ -1,43 +0,0 @@ -/* - * 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.tests.integration.io.sinks; - -import java.util.Optional; -import org.apache.pulsar.tests.integration.topologies.PulsarCluster; -import org.testcontainers.elasticsearch.ElasticsearchContainer; - -public class ElasticSearch8SinkTester extends ElasticSearchSinkTester { - - public static final String ELASTICSEARCH_8 = Optional.ofNullable(System.getenv("ELASTICSEARCH_IMAGE_V8")) - .orElse("docker.elastic.co/elasticsearch/elasticsearch:8.5.1"); - - - public ElasticSearch8SinkTester(boolean schemaEnable) { - super(schemaEnable); - } - - @Override - protected ElasticsearchContainer createSinkService(PulsarCluster cluster) { - return new ElasticsearchContainer(ELASTICSEARCH_8) - .withEnv("ES_JAVA_OPTS", "-Xms128m -Xmx256m") - .withEnv("xpack.security.enabled", "false") - .withEnv("xpack.security.http.ssl.enabled", "false"); - } - -} diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/OpenSearchSinkTester.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/OpenSearchSinkTester.java deleted file mode 100644 index 1e10cc4189c1a..0000000000000 --- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/OpenSearchSinkTester.java +++ /dev/null @@ -1,88 +0,0 @@ -/* - * 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.tests.integration.io.sinks; - -import java.util.Optional; -import org.apache.http.HttpHost; -import org.apache.pulsar.tests.integration.topologies.PulsarCluster; -import org.awaitility.Awaitility; -import org.opensearch.action.search.SearchRequest; -import org.opensearch.action.search.SearchResponse; -import org.opensearch.client.RequestOptions; -import org.opensearch.client.RestClient; -import org.opensearch.client.RestClientBuilder; -import org.opensearch.client.RestHighLevelClient; -import org.testcontainers.elasticsearch.ElasticsearchContainer; -import org.testcontainers.utility.DockerImageName; - -import java.util.Map; - -import static org.testng.Assert.assertTrue; - -public class OpenSearchSinkTester extends ElasticSearchSinkTester { - - public static final String OPENSEARCH = Optional.ofNullable(System.getenv("OPENSEARCH_IMAGE")) - .orElse("opensearchproject/opensearch:1.2.4"); - - private RestHighLevelClient elasticClient; - - - public OpenSearchSinkTester(boolean schemaEnable) { - super(schemaEnable); - } - - @Override - protected ElasticsearchContainer createSinkService(PulsarCluster cluster) { - DockerImageName dockerImageName = DockerImageName.parse(OPENSEARCH) - .asCompatibleSubstituteFor("docker.elastic.co/elasticsearch/elasticsearch"); - return new ElasticsearchContainer(dockerImageName) - .withEnv("OPENSEARCH_JAVA_OPTS", "-Xms128m -Xmx256m") - .withEnv("bootstrap.memory_lock", "true") - .withEnv("plugins.security.disabled", "true"); - } - - @Override - public void prepareSink() throws Exception { - RestClientBuilder builder = RestClient.builder( - new HttpHost( - "localhost", - serviceContainer.getMappedPort(9200), - "http")); - elasticClient = new RestHighLevelClient(builder); - } - - @Override - public void validateSinkResult(Map kvs) { - org.opensearch.action.search.SearchRequest searchRequest = new SearchRequest("test-index"); - - Awaitility.await().untilAsserted(() -> { - SearchResponse searchResult = elasticClient.search(searchRequest, RequestOptions.DEFAULT); - assertTrue(searchResult.getHits().getTotalHits().value > 0, searchResult.toString()); - }); - } - - @Override - public void close() throws Exception { - super.close(); - if (elasticClient != null) { - elasticClient.close(); - elasticClient = null; - } - } -} From 5ff244d15f7406d0e13eb4eb56bd37b34166f164 Mon Sep 17 00:00:00 2001 From: xiangying <1984997880@qq.com> Date: Sat, 25 Feb 2023 21:10:55 +0800 Subject: [PATCH 4/9] fix --- .../apache/pulsar/broker/admin/impl/PersistentTopicsBase.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) 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 1be743a7ee93b..51525303ac0af 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 @@ -5068,7 +5068,8 @@ private void internalGetReplicatedSubscriptionStatusForNonPartitionedTopic(Async // Redirect the request to the appropriate broker if this broker is not the owner of the topic validateTopicOwnership(topicName, authoritative); - getReplicatedSubscriptionStatusFromLocalBroker(topicName, subName).get(); + Map res = getReplicatedSubscriptionStatusFromLocalBroker(topicName, subName).get(); + asyncResponse.resume(res); } catch (Exception e) { log.error("[{}] Failed to get replicated subscription status on {} {}", clientAppId(), topicName, subName, e); From 056e37233f1719698b4d82426adca8993308366d Mon Sep 17 00:00:00 2001 From: xiangying <1984997880@qq.com> Date: Sat, 25 Feb 2023 21:38:06 +0800 Subject: [PATCH 5/9] fix --- .../client/admin/internal/PulsarAdminBuilderImplTest.java | 2 -- .../org/apache/pulsar/functions/instance/InstanceCache.java | 1 - .../pulsar/functions/worker/ClusterServiceCoordinator.java | 3 --- 3 files changed, 6 deletions(-) diff --git a/pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/PulsarAdminBuilderImplTest.java b/pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/PulsarAdminBuilderImplTest.java index d278a18769057..06b9c0e555b3b 100644 --- a/pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/PulsarAdminBuilderImplTest.java +++ b/pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/PulsarAdminBuilderImplTest.java @@ -34,8 +34,6 @@ import org.apache.pulsar.client.api.EncodedAuthenticationParameterSupport; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.impl.auth.AuthenticationDisabled; -import org.apache.pulsar.client.impl.conf.ClientConfigurationData; -import org.testng.Assert; import org.testng.annotations.Test; /** diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/InstanceCache.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/InstanceCache.java index 38d85335bcd52..7b73780caac8f 100644 --- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/InstanceCache.java +++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/InstanceCache.java @@ -18,7 +18,6 @@ */ package org.apache.pulsar.functions.instance; -import io.netty.util.concurrent.DefaultThreadFactory; import lombok.Getter; import java.util.concurrent.Executors; diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/ClusterServiceCoordinator.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/ClusterServiceCoordinator.java index d15d2ba90c557..5dbb61456e4d2 100644 --- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/ClusterServiceCoordinator.java +++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/ClusterServiceCoordinator.java @@ -31,9 +31,6 @@ import lombok.Setter; import lombok.extern.slf4j.Slf4j; import org.apache.pulsar.client.util.ExecutorProvider; - -import java.util.HashMap; -import java.util.Map; import java.util.function.Supplier; @Slf4j From 79771dfc342315d43f77343f0704f329720e2aa4 Mon Sep 17 00:00:00 2001 From: AloysZhang Date: Thu, 23 Feb 2023 13:18:10 +0800 Subject: [PATCH 6/9] [fix][txn]fix receive duplicated messages due to pendingAcks in PendingAckHandle (#19581) Co-authored-by: mayozhang (cherry picked from commit e6bc4999e22a763c9097494970ed6d9796278a23) --- .../pulsar/tests/MockitoCleanupListener.java | 3 +- .../mledger/util/PositionAckSetUtil.java | 7 ++ .../impl/ModularLoadManagerImpl.java | 2 +- .../service/AbstractBaseDispatcher.java | 13 ++++ .../client/impl/TransactionEndToEndTest.java | 78 +++++++++++++++++++ .../BitSetRecyclableRecyclableTest.java | 18 +++++ 6 files changed, 119 insertions(+), 2 deletions(-) diff --git a/buildtools/src/main/java/org/apache/pulsar/tests/MockitoCleanupListener.java b/buildtools/src/main/java/org/apache/pulsar/tests/MockitoCleanupListener.java index 73fff1bb7e269..dcc5707afba64 100644 --- a/buildtools/src/main/java/org/apache/pulsar/tests/MockitoCleanupListener.java +++ b/buildtools/src/main/java/org/apache/pulsar/tests/MockitoCleanupListener.java @@ -39,7 +39,8 @@ public class MockitoCleanupListener extends BetweenTestClassesListenerAdapter { protected void onBetweenTestClasses(Class endedTestClass, Class startedTestClass) { if (MOCKITO_CLEANUP_ENABLED) { if (MockitoThreadLocalStateCleaner.INSTANCE.isEnabled()) { - LOG.info("Cleaning up Mockito's ThreadSafeMockingProgress.MOCKING_PROGRESS_PROVIDER thread local state."); + LOG.info("Cleaning up Mockito's ThreadSafeMockingProgress.MOCKING_PROGRESS_PROVIDER " + + "thread local state."); MockitoThreadLocalStateCleaner.INSTANCE.cleanup(); } cleanupMockitoInline(); diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/PositionAckSetUtil.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/PositionAckSetUtil.java index da3043e7458bb..d1bc95c768a85 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/PositionAckSetUtil.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/PositionAckSetUtil.java @@ -60,6 +60,13 @@ public static long[] andAckSet(long[] firstAckSet, long[] secondAckSet) { return ackSet; } + public static boolean isAckSetEmpty(long[] ackSet) { + BitSetRecyclable bitSet = BitSetRecyclable.create().resetWords(ackSet); + boolean isEmpty = bitSet.isEmpty(); + bitSet.recycle(); + return isEmpty; + } + //This method is compare two position which position is bigger than another one. //When the ledgerId and entryId in this position is same to another one and two position all have ack set, it will //compare the ack set next bit index is bigger than another one. diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java index 1beb8ec741524..d81f6949f43f7 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java @@ -64,8 +64,8 @@ import org.apache.pulsar.broker.resources.ClusterResources; import org.apache.pulsar.broker.stats.prometheus.metrics.Summary; import org.apache.pulsar.client.admin.PulsarAdminException; -import org.apache.pulsar.common.naming.NamespaceBundle; import org.apache.pulsar.client.util.ExecutorProvider; +import org.apache.pulsar.common.naming.NamespaceBundle; import org.apache.pulsar.common.naming.NamespaceBundleFactory; import org.apache.pulsar.common.naming.NamespaceName; import org.apache.pulsar.common.naming.ServiceUnitId; diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java index 6b9ddcc1162f3..0b5108eeab861 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java @@ -20,6 +20,7 @@ package org.apache.pulsar.broker.service; import static org.apache.bookkeeper.mledger.util.PositionAckSetUtil.andAckSet; +import static org.apache.bookkeeper.mledger.util.PositionAckSetUtil.isAckSetEmpty; import com.google.common.collect.ImmutableList; import io.netty.buffer.ByteBuf; import java.util.ArrayList; @@ -246,6 +247,18 @@ public int filterEntriesForConsumer(@Nullable EntryWrapper[] entryWrapper, int e // if actSet is null, use pendingAck ackSet ackSet = positionInPendingAck.getAckSet(); } + // if the result of pendingAckSet(in pendingAckHandle) AND the ackSet(in cursor) is empty + // filter this entry + if (isAckSetEmpty(ackSet)) { + entries.set(i, null); + entry.release(); + continue; + } + } else { + // filter non-batch message in pendingAck state + entries.set(i, null); + entry.release(); + continue; } } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java index b0a4b28bbdc84..a51ad6998ed11 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java @@ -171,6 +171,84 @@ private void testIndividualAckAbortFilterAckSetInPendingAckState() throws Except assertNull(consumer.receive(2, TimeUnit.SECONDS)); } + + @Test(dataProvider="enableBatch") + private void testFilterMsgsInPendingAckStateWhenConsumerDisconnect(boolean enableBatch) throws Exception { + final String topicName = NAMESPACE1 + "/testFilterMsgsInPendingAckStateWhenConsumerDisconnect-" + enableBatch; + final int count = 10; + + @Cleanup + Producer producer = null; + if (enableBatch) { + producer = pulsarClient + .newProducer(Schema.INT32) + .topic(topicName) + .enableBatching(true) + .batchingMaxPublishDelay(1, TimeUnit.HOURS) + .batchingMaxMessages(count).create(); + } else { + producer = pulsarClient + .newProducer(Schema.INT32) + .topic(topicName) + .enableBatching(false).create(); + } + + @Cleanup + Consumer consumer = pulsarClient + .newConsumer(Schema.INT32) + .topic(topicName) + .isAckReceiptEnabled(true) + .subscriptionName("test") + .subscriptionType(SubscriptionType.Shared) + .enableBatchIndexAcknowledgment(true) + .subscribe(); + + for (int i = 0; i < count; i++) { + producer.sendAsync(i); + } + + Transaction txn1 = getTxn(); + + Transaction txn2 = getTxn(); + + + // txn1 ack half of messages and don't end the txn1 + for (int i = 0; i < count / 2; i++) { + consumer.acknowledgeAsync(consumer.receive().getMessageId(), txn1).get(); + } + + // txn2 ack the rest half of messages and commit tnx2 + for (int i = count / 2; i < count; i++) { + consumer.acknowledgeAsync(consumer.receive().getMessageId(), txn2).get(); + } + // commit txn2 + txn2.commit().get(); + + // close and re-create consumer + consumer.close(); + consumer = pulsarClient + .newConsumer(Schema.INT32) + .topic(topicName) + .isAckReceiptEnabled(true) + .subscriptionName("test") + .subscriptionType(SubscriptionType.Shared) + .enableBatchIndexAcknowledgment(true) + .subscribe(); + + Message message = consumer.receive(3, TimeUnit.SECONDS); + Assert.assertNull(message); + + // abort txn1 + txn1.abort().get(); + // after txn1 aborted, consumer will receive messages txn1 contains + int receiveCounter = 0; + while((message = consumer.receive(3, TimeUnit.SECONDS)) != null) { + Assert.assertEquals(message.getValue().intValue(), receiveCounter); + receiveCounter ++; + } + Assert.assertEquals(receiveCounter, count / 2); + } + @Test(dataProvider="enableBatch") private void produceCommitTest(boolean enableBatch) throws Exception { @Cleanup diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/BitSetRecyclableRecyclableTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/BitSetRecyclableRecyclableTest.java index 0f42f35608a76..b1a94c4916224 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/BitSetRecyclableRecyclableTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/BitSetRecyclableRecyclableTest.java @@ -45,4 +45,22 @@ public void testResetWords() { Assert.assertTrue(bitset1.get(128)); Assert.assertFalse(bitset1.get(256)); } + + @Test + public void testBitSetEmpty() { + BitSetRecyclable bitSet = BitSetRecyclable.create(); + bitSet.set(0, 5); + bitSet.clear(1); + bitSet.clear(2); + bitSet.clear(3); + long[] array = bitSet.toLongArray(); + Assert.assertFalse(bitSet.isEmpty()); + Assert.assertFalse(BitSetRecyclable.create().resetWords(array).isEmpty()); + bitSet.clear(0); + bitSet.clear(4); + Assert.assertTrue(bitSet.isEmpty()); + long[] array1 = bitSet.toLongArray(); + Assert.assertTrue(BitSetRecyclable.create().resetWords(array1).isEmpty()); + bitSet.recycle(); + } } From ab8af950cfd1d4e43f8cffe46a342b16b00e44ac Mon Sep 17 00:00:00 2001 From: xiangying <1984997880@qq.com> Date: Sun, 26 Feb 2023 15:12:14 +0800 Subject: [PATCH 7/9] Revert "[fix][test] Fix flaky test deleteNamespaceGracefully (#18220)" This reverts commit a3e593abbcb3a88509816ccf584823478f21aa74. --- .../pulsar/broker/admin/AdminApi2Test.java | 8 +- .../pulsar/broker/admin/AdminApiTest.java | 38 ++--- .../broker/admin/AdminApiTlsAuthTest.java | 2 +- .../auth/MockedPulsarServiceBaseTest.java | 142 ------------------ .../NamespaceOwnershipListenerTests.java | 13 +- .../OwnerShipForCurrentServerTestBase.java | 3 +- .../ResourceGroupConfigListenerTest.java | 2 +- .../service/BacklogQuotaManagerTest.java | 25 +-- ...erServiceBundlesCacheInvalidationTest.java | 2 +- .../service/PersistentTopicE2ETest.java | 4 +- .../transaction/TransactionTestBase.java | 21 +-- .../api/DispatcherBlockConsumerTest.java | 3 +- 12 files changed, 42 insertions(+), 221 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java index 3c99452fdc07d..fcc2f14490784 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java @@ -1386,7 +1386,7 @@ public void testDeleteTenant() throws Exception { assertTrue(admin.topics().getList(namespace).isEmpty()); // delete namespace - deleteNamespaceWithRetry(namespace, false); + admin.namespaces().deleteNamespace(namespace, false); assertFalse(admin.namespaces().getNamespaces(tenant).contains(namespace)); assertTrue(admin.namespaces().getNamespaces(tenant).isEmpty()); @@ -1475,7 +1475,7 @@ public void testDeleteNamespace(NamespaceAttr namespaceAttr) throws Exception { admin.topics().deletePartitionedTopic(topic); // delete namespace - deleteNamespaceWithRetry(namespace, false); + admin.namespaces().deleteNamespace(namespace, false); assertFalse(admin.namespaces().getNamespaces(tenant).contains(namespace)); assertTrue(admin.namespaces().getNamespaces(tenant).isEmpty()); @@ -1572,7 +1572,7 @@ public void testDeleteNamespaceWithTopicPolicies() throws Exception { }); producer.close(); admin.topics().delete(topic); - deleteNamespaceWithRetry(namespace, false); + admin.namespaces().deleteNamespace(namespace); Awaitility.await().untilAsserted(() -> { assertTrue(admin.namespaces().getNamespaces(tenant).isEmpty()); }); @@ -1783,7 +1783,7 @@ public void testForceDeleteNamespace() throws Exception { final String topic = "persistent://" + namespaceName + "/test" + UUID.randomUUID(); pulsarClient.newProducer(Schema.DOUBLE).topic(topic).create().close(); Awaitility.await().untilAsserted(() -> assertNotNull(admin.schemas().getSchemaInfo(topic))); - deleteNamespaceWithRetry(namespaceName, true); + admin.namespaces().deleteNamespace(namespaceName, true); try { admin.schemas().getSchemaInfo(topic); } catch (PulsarAdminException e) { 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 a9ba361bbb9ca..3f151daf8d207 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 @@ -196,7 +196,7 @@ public void resetClusters() throws Exception { pulsar.getConfiguration().setForceDeleteNamespaceAllowed(true); for (String tenant : admin.tenants().getTenants()) { for (String namespace : admin.namespaces().getNamespaces(tenant)) { - deleteNamespaceWithRetry(namespace, true); + deleteNamespaceGraceFully(namespace, true); } admin.tenants().deleteTenant(tenant, true); } @@ -291,7 +291,7 @@ public void clusters() throws Exception { Awaitility.await() .untilAsserted(() -> assertEquals(admin.clusters().getClusters(), Lists.newArrayList("test"))); - deleteNamespaceWithRetry("prop-xyz/ns1", false); + admin.namespaces().deleteNamespace("prop-xyz/ns1"); admin.clusters().deleteCluster("test"); assertEquals(admin.clusters().getClusters(), Lists.newArrayList()); @@ -516,7 +516,7 @@ public void brokers() throws Exception { String.format("%s:%d", parts[0], pulsar.getListenPortHTTPS().get())); Assert.assertEquals(nsMap2.size(), 2); - deleteNamespaceWithRetry("prop-xyz/ns1", false); + admin.namespaces().deleteNamespace("prop-xyz/ns1"); admin.clusters().deleteCluster("test"); assertEquals(admin.clusters().getClusters(), Lists.newArrayList()); } @@ -710,7 +710,7 @@ public void testGetDynamicLocalConfiguration() throws Exception { } @Test - public void properties() throws Exception { + public void properties() throws PulsarAdminException { try { admin.tenants().getTenantInfo("does-not-exist"); fail("should have failed"); @@ -739,7 +739,7 @@ public void properties() throws Exception { assertEquals(e.getStatusCode(), 409); assertEquals(e.getMessage(), "The tenant still has active namespaces"); } - deleteNamespaceWithRetry("prop-xyz/ns1", false); + admin.namespaces().deleteNamespace("prop-xyz/ns1"); admin.tenants().deleteTenant("prop-xyz"); assertEquals(admin.tenants().getTenants(), Lists.newArrayList()); @@ -768,7 +768,7 @@ public void namespaces() throws Exception { assertEquals(admin.namespaces().getPolicies("prop-xyz/ns3").bundles.getNumBundles(), 4); assertEquals(admin.namespaces().getPolicies("prop-xyz/ns3").bundles.getBoundaries().size(), 5); - deleteNamespaceWithRetry("prop-xyz/ns3", false); + admin.namespaces().deleteNamespace("prop-xyz/ns3"); try { admin.namespaces().createNamespace("non-existing/ns1"); @@ -842,7 +842,7 @@ public void namespaces() throws Exception { } assertTrue(i < 10); - deleteNamespaceWithRetry("prop-xyz/ns1", false); + admin.namespaces().deleteNamespace("prop-xyz/ns1"); assertEquals(admin.namespaces().getNamespaces("prop-xyz"), Lists.newArrayList("prop-xyz/ns2")); try { @@ -1266,7 +1266,7 @@ public void testGetPartitionedStatsInternal() throws Exception { @Test(dataProvider = "numBundles") public void testDeleteNamespaceBundle(Integer numBundles) throws Exception { - deleteNamespaceWithRetry("prop-xyz/ns1", false); + admin.namespaces().deleteNamespace("prop-xyz/ns1"); admin.namespaces().createNamespace("prop-xyz/ns1-bundles", numBundles); admin.namespaces().setNamespaceReplicationClusters("prop-xyz/ns1-bundles", Sets.newHashSet("test")); @@ -1278,8 +1278,8 @@ public void testDeleteNamespaceBundle(Integer numBundles) throws Exception { assertEquals(admin.namespaces().getTopics("prop-xyz/ns1-bundles"), Lists.newArrayList()); - deleteNamespaceWithRetry("prop-xyz/ns1-bundles", false); - assertEquals(admin.namespaces().getNamespaces("prop-xyz", "test"), new ArrayList<>()); + admin.namespaces().deleteNamespace("prop-xyz/ns1-bundles"); + assertEquals(admin.namespaces().getNamespaces("prop-xyz", "test"), Lists.newArrayList()); } @Test @@ -1372,7 +1372,7 @@ public void testDeleteNamespaceForcefully() throws Exception { } // delete namespace forcefully - deleteNamespaceWithRetry(namespace, true); + admin.namespaces().deleteNamespace(namespace, true); assertFalse(admin.namespaces().getNamespaces(tenant).contains(namespace)); assertTrue(admin.namespaces().getNamespaces(tenant).isEmpty()); @@ -1432,7 +1432,7 @@ public void testForceDeleteTenantNotAllowed() throws Exception { @Test public void testNamespaceSplitBundle() throws Exception { - admin.namespaces().createNamespace("prop-xyz/splitBundle", Sets.newHashSet("test")); + admin.namespaces().createNamespace("prop-xyz/splitBundle", Set.of("test")); // Force to create a topic final String namespace = "prop-xyz/splitBundle"; @@ -1656,7 +1656,7 @@ public void testNamespaceSplitBundleConcurrent() throws Exception { @Test public void testNamespaceUnloadBundle() throws Exception { - admin.namespaces().createNamespace("prop-xyz/unloadBundle", Sets.newHashSet("test")); + admin.namespaces().createNamespace("prop-xyz/unloadBundle", Set.of("test")); assertEquals(admin.topics().getList("prop-xyz/unloadBundle"), new ArrayList<>()); @@ -2086,7 +2086,7 @@ public void testBackwardCompatibility() throws Exception { assertEquals(result.someNewIntField, 0); assertNull(result.someNewString); - deleteNamespaceWithRetry("prop-xyz/ns1", false); + admin.namespaces().deleteNamespace("prop-xyz/ns1"); admin.tenants().deleteTenant("prop-xyz"); assertEquals(admin.tenants().getTenants(), Lists.newArrayList()); } @@ -2974,9 +2974,9 @@ public void testSubscriptionExpiry() throws Exception { admin.topics().delete(topic1); admin.topics().delete(topic2); admin.topics().delete(topic3); - deleteNamespaceWithRetry(namespace1, false); - deleteNamespaceWithRetry(namespace2, false); - deleteNamespaceWithRetry(namespace3, false); + admin.namespaces().deleteNamespace(namespace1); + admin.namespaces().deleteNamespace(namespace2); + admin.namespaces().deleteNamespace(namespace3); } @Test @@ -2989,11 +2989,11 @@ public void testCreateAndDeleteNamespaceWithBundles() throws Exception { String ns = BrokerTestUtil.newUniqueName("prop-xyz/ns"); admin.namespaces().createNamespace(ns, 24); - deleteNamespaceWithRetry(ns, false); + admin.namespaces().deleteNamespace(ns); // Re-create and re-delete admin.namespaces().createNamespace(ns, 32); - deleteNamespaceWithRetry(ns, false); + admin.namespaces().deleteNamespace(ns); } @Test diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTlsAuthTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTlsAuthTest.java index 11b806a656fdb..311e31be735da 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTlsAuthTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTlsAuthTest.java @@ -443,7 +443,7 @@ public void testDeleteNamespace() throws Exception { admin.topics().delete("tenant1/ns1/foobar", true); log.info("Deleting namespace"); - deleteNamespaceWithRetry("tenant1/ns1", false, admin); + admin.namespaces().deleteNamespace("tenant1/ns1"); } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java index 4f3af27a38449..84964680d3472 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java @@ -33,20 +33,13 @@ import java.net.URL; import java.nio.charset.StandardCharsets; import java.util.ArrayList; -import java.util.Arrays; -import java.util.Collection; -import java.util.Date; import java.util.List; import java.util.Map; import java.util.Optional; import java.util.Set; -import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; -import java.util.function.Consumer; import java.util.function.Predicate; import java.util.function.Supplier; -import javax.ws.rs.container.AsyncResponse; -import javax.ws.rs.container.TimeoutHandler; import org.apache.bookkeeper.client.BookKeeper; import org.apache.bookkeeper.client.EnsemblePlacementPolicy; import org.apache.bookkeeper.client.PulsarMockBookKeeper; @@ -58,7 +51,6 @@ import org.apache.pulsar.broker.ServiceConfiguration; import org.apache.pulsar.broker.intercept.CounterBrokerInterceptor; import org.apache.pulsar.broker.namespace.NamespaceService; -import org.apache.pulsar.broker.service.BrokerTestBase; import org.apache.pulsar.client.admin.PulsarAdmin; import org.apache.pulsar.client.admin.PulsarAdminBuilder; import org.apache.pulsar.client.admin.PulsarAdminException; @@ -75,7 +67,6 @@ import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.MockZooKeeper; import org.apache.zookeeper.data.ACL; -import org.awaitility.Awaitility; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.testng.annotations.DataProvider; @@ -507,139 +498,6 @@ protected void setupDefaultTenantAndNamespace() throws Exception { } } - protected Object asyncRequests(Consumer function) throws Exception { - TestAsyncResponse ctx = new TestAsyncResponse(); - function.accept(ctx); - ctx.latch.await(); - if (ctx.e != null) { - throw (Exception) ctx.e; - } - return ctx.response; - } - - public static class TestAsyncResponse implements AsyncResponse { - - Object response; - Throwable e; - CountDownLatch latch = new CountDownLatch(1); - - @Override - public boolean resume(Object response) { - this.response = response; - latch.countDown(); - return true; - } - - @Override - public boolean resume(Throwable response) { - this.e = response; - latch.countDown(); - return true; - } - - @Override - public boolean cancel() { - return false; - } - - @Override - public boolean cancel(int retryAfter) { - return false; - } - - @Override - public boolean cancel(Date retryAfter) { - return false; - } - - @Override - public boolean isSuspended() { - return false; - } - - @Override - public boolean isCancelled() { - return false; - } - - @Override - public boolean isDone() { - return false; - } - - @Override - public boolean setTimeout(long time, TimeUnit unit) { - return false; - } - - @Override - public void setTimeoutHandler(TimeoutHandler handler) { - - } - - @Override - public Collection> register(Class callback) { - return null; - } - - @Override - public Map, Collection>> register(Class callback, Class... callbacks) { - return null; - } - - @Override - public Collection> register(Object callback) { - return null; - } - - @Override - public Map, Collection>> register(Object callback, Object... callbacks) { - return null; - } - - } - - /** - * see {@link BrokerTestBase#deleteNamespaceWithRetry(String, boolean, PulsarAdmin, Collection)} - */ - protected void deleteNamespaceWithRetry(String ns, boolean force) - throws Exception { - BrokerTestBase.deleteNamespaceWithRetry(ns, force, admin, pulsar); - } - - /** - * see {@link BrokerTestBase#deleteNamespaceWithRetry(String, boolean, PulsarAdmin, Collection)} - */ - protected void deleteNamespaceWithRetry(String ns, boolean force, PulsarAdmin admin) - throws Exception { - BrokerTestBase.deleteNamespaceWithRetry(ns, force, admin, pulsar); - } - - /** - * see {@link MockedPulsarServiceBaseTest#deleteNamespaceWithRetry(String, boolean, PulsarAdmin, Collection)} - */ - public static void deleteNamespaceWithRetry(String ns, boolean force, PulsarAdmin admin, PulsarService...pulsars) - throws Exception { - deleteNamespaceWithRetry(ns, force, admin, Arrays.asList(pulsars)); - } - - /** - * 1. Pause system "__change_event" topic creates. - * 2. Do delete namespace with retry because maybe fail by race-condition with create topics. - */ - public static void deleteNamespaceWithRetry(String ns, boolean force, PulsarAdmin admin, - Collection pulsars) throws Exception { - Awaitility.await().atMost(5, TimeUnit.SECONDS).until(() -> { - try { - // Maybe fail by race-condition with create topics, just retry. - admin.namespaces().deleteNamespace(ns, force); - return true; - } catch (Exception ex) { - return false; - } - }); - } - @DataProvider(name = "invalidPersistentPolicies") public Object[][] incorrectPersistentPolicies() { return new Object[][] { diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceOwnershipListenerTests.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceOwnershipListenerTests.java index bf0e252660a71..6a57307c7a142 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceOwnershipListenerTests.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceOwnershipListenerTests.java @@ -52,7 +52,7 @@ protected void cleanup() throws Exception { } @Test - public void testNamespaceBundleOwnershipListener() throws Exception { + public void testNamespaceBundleOwnershipListener() throws PulsarAdminException, InterruptedException, PulsarClientException { final CountDownLatch countDownLatch = new CountDownLatch(2); final AtomicBoolean onLoad = new AtomicBoolean(false); @@ -98,11 +98,11 @@ public void unLoad(NamespaceBundle bundle) { Assert.assertTrue(onLoad.get()); Assert.assertTrue(unLoad.get()); admin.topics().delete(topic); - deleteNamespaceWithRetry(namespace, false); + admin.namespaces().deleteNamespace(namespace); } @Test - public void testGetAllPartitions() throws Exception { + public void testGetAllPartitions() throws PulsarAdminException, ExecutionException, InterruptedException { final String namespace = "prop/" + UUID.randomUUID().toString(); admin.namespaces().createNamespace(namespace, Sets.newHashSet("test")); assertTrue(admin.namespaces().getNamespaces("prop").contains(namespace)); @@ -119,11 +119,12 @@ public void testGetAllPartitions() throws Exception { } admin.topics().deletePartitionedTopic(topicName); - deleteNamespaceWithRetry(namespace, false); + admin.namespaces().deleteNamespace(namespace); } @Test - public void testNamespaceBundleLookupOnwershipListener() throws Exception { + public void testNamespaceBundleLookupOnwershipListener() throws PulsarAdminException, InterruptedException, + PulsarClientException { final CountDownLatch countDownLatch = new CountDownLatch(2); final AtomicInteger onLoad = new AtomicInteger(0); final AtomicInteger unLoad = new AtomicInteger(0); @@ -168,6 +169,6 @@ public boolean test(NamespaceBundle namespaceBundle) { Assert.assertEquals(onLoad.get(), 1); Assert.assertEquals(unLoad.get(), 1); admin.topics().delete(topic); - deleteNamespaceWithRetry(namespace, false); + admin.namespaces().deleteNamespace(namespace); } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/OwnerShipForCurrentServerTestBase.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/OwnerShipForCurrentServerTestBase.java index 80558361f4be8..c25c6c23fb011 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/OwnerShipForCurrentServerTestBase.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/OwnerShipForCurrentServerTestBase.java @@ -130,8 +130,7 @@ protected void setupBrokerMocks(PulsarService pulsar) throws Exception { MockZooKeeperSession mockZooKeeperSession = MockZooKeeperSession.newInstance(mockZooKeeper); doReturn(new ZKMetadataStore(mockZooKeeperSession)).when(pulsar).createLocalMetadataStore(); doReturn(new ZKMetadataStore(mockZooKeeperSession)).when(pulsar).createConfigurationMetadataStore(); - Supplier namespaceServiceSupplier = () -> spyWithClassAndConstructorArgs( - NamespaceService.class, pulsar); + Supplier namespaceServiceSupplier = () -> spyWithClassAndConstructorArgs(NamespaceService.class, pulsar); doReturn(namespaceServiceSupplier).when(pulsar).getNamespaceServiceProvider(); SameThreadOrderedSafeExecutor executor = new SameThreadOrderedSafeExecutor(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupConfigListenerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupConfigListenerTest.java index 64c5fee14d057..38cae3253fe33 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupConfigListenerTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupConfigListenerTest.java @@ -197,7 +197,7 @@ public void testResourceGroupAttachToNamespace() throws Exception { assertNull(pulsar.getResourceGroupServiceManager() .getNamespaceResourceGroup(NamespaceName.get(namespaceName)))); - deleteNamespaceWithRetry(namespaceName, false); + admin.namespaces().deleteNamespace(namespaceName); deleteResourceGroup(rgName); } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BacklogQuotaManagerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BacklogQuotaManagerTest.java index 96fc452745bf7..1d2878ca64471 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BacklogQuotaManagerTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BacklogQuotaManagerTest.java @@ -40,7 +40,6 @@ import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl; import org.apache.pulsar.broker.PulsarService; import org.apache.pulsar.broker.ServiceConfiguration; -import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest; import org.apache.pulsar.broker.service.persistent.PersistentTopic; import org.apache.pulsar.client.admin.GetStatsOptions; import org.apache.pulsar.client.admin.PulsarAdmin; @@ -83,22 +82,6 @@ public class BacklogQuotaManagerTest { private static final int TIME_TO_CHECK_BACKLOG_QUOTA = 2; private static final int MAX_ENTRIES_PER_LEDGER = 5; - /** - * see {@link MockedPulsarServiceBaseTest#deleteNamespaceWithRetry(String, boolean, PulsarAdmin, Collection)} - */ - protected void deleteNamespaceWithRetry(String ns, boolean force) - throws Exception { - MockedPulsarServiceBaseTest.deleteNamespaceWithRetry(ns, force, admin, pulsar); - } - - /** - * see {@link MockedPulsarServiceBaseTest#deleteNamespaceWithRetry(String, boolean, PulsarAdmin, Collection)} - */ - protected void deleteNamespaceWithRetry(String ns, boolean force, PulsarAdmin admin) - throws Exception { - MockedPulsarServiceBaseTest.deleteNamespaceWithRetry(ns, force, admin, pulsar); - } - @DataProvider(name = "backlogQuotaSizeGB") public Object[][] backlogQuotaSizeGB() { return new Object[][] { { true }, { false } }; @@ -175,10 +158,10 @@ void createNamespaces() throws PulsarAdminException { } @AfterMethod(alwaysRun = true) - void clearNamespaces() throws Exception { - deleteNamespaceWithRetry("prop/ns-quota", true); - deleteNamespaceWithRetry("prop/quotahold", true); - deleteNamespaceWithRetry("prop/quotaholdasync", true); + void clearNamespaces() throws PulsarAdminException { + admin.namespaces().deleteNamespace("prop/ns-quota", true); + admin.namespaces().deleteNamespace("prop/quotahold", true); + admin.namespaces().deleteNamespace("prop/quotaholdasync", true); } private void rolloverStats() { diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceBundlesCacheInvalidationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceBundlesCacheInvalidationTest.java index a55540c3f4f9d..573d473ed587a 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceBundlesCacheInvalidationTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceBundlesCacheInvalidationTest.java @@ -59,7 +59,7 @@ public void testRecreateNamespace() throws Exception { // Delete and recreate with 32 bundles admin.topics().delete(topic); - deleteNamespaceWithRetry(namespace, false); + admin.namespaces().deleteNamespace(namespace, false); admin.namespaces().createNamespace(namespace, 32); BundlesData bundlesData = admin.namespaces().getBundles(namespace); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicE2ETest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicE2ETest.java index fdd1170c21b38..54dc706016bbf 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicE2ETest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicE2ETest.java @@ -958,7 +958,7 @@ public void testMessageExpiry() throws Exception { consumer.close(); admin.topics().deleteSubscription(topicName, subName); admin.topics().delete(topicName); - deleteNamespaceWithRetry(namespaceName, false); + admin.namespaces().deleteNamespace(namespaceName); } @Test @@ -1052,7 +1052,7 @@ public void testMessageExpiryWithTopicMessageTTL() throws Exception { consumer.close(); admin.topics().deleteSubscription(topicName, subName); admin.topics().delete(topicName); - deleteNamespaceWithRetry(namespaceName, true); + admin.namespaces().deleteNamespace(namespaceName, true); } catch (PulsarAdminException e) { Assert.assertEquals(e.getStatusCode(), 500); } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTestBase.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTestBase.java index 990f63011b320..25c555f09b949 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTestBase.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTestBase.java @@ -44,11 +44,9 @@ import org.apache.pulsar.broker.BookKeeperClientFactory; import org.apache.pulsar.broker.PulsarService; import org.apache.pulsar.broker.ServiceConfiguration; -import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest; import org.apache.pulsar.broker.auth.SameThreadOrderedSafeExecutor; import org.apache.pulsar.broker.intercept.CounterBrokerInterceptor; import org.apache.pulsar.broker.namespace.NamespaceService; -import org.apache.pulsar.broker.service.BrokerTestBase; import org.apache.pulsar.client.admin.PulsarAdmin; import org.apache.pulsar.client.admin.PulsarAdminBuilder; import org.apache.pulsar.client.api.ClientBuilder; @@ -200,8 +198,7 @@ protected void setupBrokerMocks(PulsarService pulsar) throws Exception { MockZooKeeperSession mockZooKeeperSession = MockZooKeeperSession.newInstance(mockZooKeeper); doReturn(new ZKMetadataStore(mockZooKeeperSession)).when(pulsar).createLocalMetadataStore(); doReturn(new ZKMetadataStore(mockZooKeeperSession)).when(pulsar).createConfigurationMetadataStore(); - Supplier namespaceServiceSupplier = - () -> spyWithClassAndConstructorArgs(NamespaceService.class, pulsar); + Supplier namespaceServiceSupplier = () -> spyWithClassAndConstructorArgs(NamespaceService.class, pulsar); doReturn(namespaceServiceSupplier).when(pulsar).getNamespaceServiceProvider(); SameThreadOrderedSafeExecutor executor = new SameThreadOrderedSafeExecutor(); @@ -333,20 +330,4 @@ protected final void internalCleanup() { log.warn("Failed to clean up mocked pulsar service:", e); } } - - /** - * see {@link BrokerTestBase#deleteNamespaceWithRetry(String, boolean, PulsarAdmin, Collection)} - */ - protected void deleteNamespaceWithRetry(String ns, boolean force) - throws Exception { - MockedPulsarServiceBaseTest.deleteNamespaceWithRetry(ns, force, admin, pulsarServiceList); - } - - /** - * see {@link MockedPulsarServiceBaseTest#deleteNamespaceWithRetry(String, boolean, PulsarAdmin, Collection)} - */ - protected void deleteNamespaceWithRetry(String ns, boolean force, PulsarAdmin admin) - throws Exception { - MockedPulsarServiceBaseTest.deleteNamespaceWithRetry(ns, force, admin, pulsarServiceList); - } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DispatcherBlockConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DispatcherBlockConsumerTest.java index df695cbf715a8..1a56d689799be 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DispatcherBlockConsumerTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DispatcherBlockConsumerTest.java @@ -631,8 +631,7 @@ public void testBrokerSubscriptionRecovery(boolean unloadBundleGracefully) throw // if broker unload bundle gracefully then cursor metadata recovered from zk else from ledger if (unloadBundleGracefully) { // set clean namespace which will not let broker unload bundle gracefully: stop broker - Supplier namespaceServiceSupplier = - () -> spyWithClassAndConstructorArgs(NamespaceService.class, pulsar); + Supplier namespaceServiceSupplier = () -> spyWithClassAndConstructorArgs(NamespaceService.class, pulsar); doReturn(namespaceServiceSupplier).when(pulsar).getNamespaceServiceProvider(); } stopBroker(); From 57e81b2716d3c99de4a93d1359e665c78f8f9372 Mon Sep 17 00:00:00 2001 From: xiangying <1984997880@qq.com> Date: Sun, 26 Feb 2023 15:12:27 +0800 Subject: [PATCH 8/9] Revert "[improve][test] Improve AdminApiTest to reduce the execution time (#17980)" This reverts commit de786f632798eddc1ec4c8c65c622884ae39d9a4. --- .../pulsar/broker/admin/AdminApiTest.java | 131 +++++------------- 1 file changed, 31 insertions(+), 100 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 3f151daf8d207..7206de3609072 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 @@ -135,9 +135,8 @@ 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.BeforeClass; +import org.testng.annotations.BeforeMethod; import org.testng.annotations.DataProvider; import org.testng.annotations.Test; @@ -160,7 +159,7 @@ public class AdminApiTest extends MockedPulsarServiceBaseTest { private NamespaceBundleFactory bundleFactory; - @BeforeClass + @BeforeMethod @Override public void setup() throws Exception { conf.setLoadBalancerEnabled(true); @@ -187,39 +186,14 @@ public void setup() throws Exception { otherPulsar = mockPulsarSetup.getPulsar(); otheradmin = mockPulsarSetup.getAdmin(); - 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 { + // Setup namespaces admin.clusters().createCluster("test", ClusterData.builder().serviceUrl(pulsar.getWebServiceAddress()).build()); TenantInfoImpl tenantInfo = new TenantInfoImpl(Sets.newHashSet("role1", "role2"), Sets.newHashSet("test")); admin.tenants().createTenant("prop-xyz", tenantInfo); admin.namespaces().createNamespace("prop-xyz/ns1", Sets.newHashSet("test")); } - @AfterClass(alwaysRun = true) + @AfterMethod(alwaysRun = true) @Override public void cleanup() throws Exception { adminTls.close(); @@ -1193,7 +1167,6 @@ 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); @@ -1231,9 +1204,7 @@ public void testGetPartitionedStatsInternal() throws Exception { assertEquals(admin.topics().getPartitionedTopicMetadata(partitionedTopicName).partitions, 2); // create consumer and subscription - @Cleanup - Consumer consumer = - pulsarClient.newConsumer().topic(partitionedTopicName).subscriptionName(subName).subscribe(); + pulsarClient.newConsumer().topic(partitionedTopicName).subscriptionName(subName).subscribe(); // publish several messages publishMessagesOnPersistentTopic(partitionedTopicName, 10); @@ -1432,10 +1403,8 @@ 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/splitBundle"; + final String namespace = "prop-xyz/ns1"; final String topicName = (new StringBuilder("persistent://")).append(namespace).append("/ds2").toString(); Producer producer = pulsarClient.newProducer(Schema.BYTES) .topic(topicName) @@ -1497,9 +1466,7 @@ public void testNamespaceSplitBundleWithTopicCountEquallyDivideAlgorithm() throw for (int i = 0; i < bundles.getBundles().size(); i++) { assertNotEquals(bundles.getBundles().get(i).toString(), splitRange[i]); } - for (Producer producer : producers) { - producer.close(); - } + producers.forEach(Producer::closeAsync); } @Test @@ -1516,9 +1483,6 @@ 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"; @@ -1552,9 +1516,7 @@ public void testNamespaceSplitBundleWithDefaultTopicCountEquallyDivideAlgorithm( for (int i = 0; i < bundles.getBundles().size(); i++) { assertNotEquals(bundles.getBundles().get(i).toString(), splitRange[i]); } - for (Producer producer : producers) { - producer.close(); - } + producers.forEach(Producer::closeAsync); conf.setDefaultNamespaceBundleSplitAlgorithm(NamespaceBundleSplitAlgorithm.RANGE_EQUALLY_DIVIDE_NAME); } @@ -1656,24 +1618,22 @@ public void testNamespaceSplitBundleConcurrent() throws Exception { @Test public void testNamespaceUnloadBundle() throws Exception { - admin.namespaces().createNamespace("prop-xyz/unloadBundle", Set.of("test")); - - assertEquals(admin.topics().getList("prop-xyz/unloadBundle"), new ArrayList<>()); + assertEquals(admin.topics().getList("prop-xyz/ns1"), Lists.newArrayList()); // Force to create a topic - publishMessagesOnPersistentTopic("persistent://prop-xyz/unloadBundle/ds2", 0); - assertEquals(admin.topics().getList("prop-xyz/unloadBundle"), - Lists.newArrayList("persistent://prop-xyz/unloadBundle/ds2")); + publishMessagesOnPersistentTopic("persistent://prop-xyz/ns1/ds2", 0); + assertEquals(admin.topics().getList("prop-xyz/ns1"), + Lists.newArrayList("persistent://prop-xyz/ns1/ds2")); // create consumer and subscription - Consumer consumer = pulsarClient.newConsumer().topic("persistent://prop-xyz/unloadBundle/ds2") + Consumer consumer = pulsarClient.newConsumer().topic("persistent://prop-xyz/ns1/ds2") .subscriptionName("my-sub").subscribe(); - assertEquals(admin.topics().getSubscriptions("persistent://prop-xyz/unloadBundle/ds2"), + assertEquals(admin.topics().getSubscriptions("persistent://prop-xyz/ns1/ds2"), Lists.newArrayList("my-sub")); // Create producer Producer producer = pulsarClient.newProducer(Schema.BYTES) - .topic("persistent://prop-xyz/unloadBundle/ds2") + .topic("persistent://prop-xyz/ns1/ds2") .enableBatching(false) .messageRoutingMode(MessageRoutingMode.SinglePartition) .create(); @@ -1686,13 +1646,13 @@ public void testNamespaceUnloadBundle() throws Exception { producer.close(); try { - admin.namespaces().unloadNamespaceBundle("prop-xyz/unloadBundle", "0x00000000_0xffffffff"); + admin.namespaces().unloadNamespaceBundle("prop-xyz/ns1", "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/unloadBundle"), + NamespaceBundle bundle = bundleFactory.getBundle(NamespaceName.get("prop-xyz/ns1"), Range.range(0L, BoundType.CLOSED, 0xffffffffL, BoundType.CLOSED)); assertFalse(pulsar.getNamespaceService().isServiceUnitOwned(bundle)); assertFalse(otherPulsar.getNamespaceService().isServiceUnitOwned(bundle)); @@ -1702,17 +1662,14 @@ 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/unloadBundle/ds2") != null); + .until(() -> admin.topics().getStats("persistent://prop-xyz/ns1/ds2") != null); - admin.topics().deleteSubscription("persistent://prop-xyz/unloadBundle/ds2", "my-sub"); - admin.topics().delete("persistent://prop-xyz/unloadBundle/ds2"); + admin.topics().deleteSubscription("persistent://prop-xyz/ns1/ds2", "my-sub"); + admin.topics().delete("persistent://prop-xyz/ns1/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", Sets.newHashSet("test")); @@ -1818,25 +1775,15 @@ public void testClearBacklogOnNamespace(Integer numBundles) throws Exception { admin.namespaces().setNamespaceReplicationClusters("prop-xyz/ns1-bundles", Sets.newHashSet("test")); // create consumer and subscription - @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") + pulsarClient.newConsumer().topic("persistent://prop-xyz/ns1-bundles/ds2").subscriptionName("my-sub") .subscribe(); - @Cleanup - Consumer consumer4 = - pulsarClient.newConsumer().topic("persistent://prop-xyz/ns1-bundles/ds1").subscriptionName("my-sub") + pulsarClient.newConsumer().topic("persistent://prop-xyz/ns1-bundles/ds2").subscriptionName("my-sub-1") .subscribe(); - @Cleanup - Consumer consumer5 = - pulsarClient.newConsumer().topic("persistent://prop-xyz/ns1-bundles/ds1").subscriptionName("my-sub-1") + 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") + .subscribe(); + pulsarClient.newConsumer().topic("persistent://prop-xyz/ns1-bundles/ds1").subscriptionName("my-sub-1") .subscribe(); // Create producer @@ -1900,8 +1847,7 @@ 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(); - @Cleanup - Consumer consumer3 = pulsarClient.newConsumer().topic("persistent://prop-xyz/ns1-bundles/ds2") + /* 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(); @@ -2660,9 +2606,7 @@ 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 - @Cleanup - Consumer consumer = - pulsarClient.newConsumer().topic(topic1).subscriptionName("my-subscriber-name").subscribe(); + pulsarClient.newConsumer().topic(topic1).subscriptionName("my-subscriber-name").subscribe(); TopicsImpl persistent = (TopicsImpl) admin.topics(); Field field = TopicsImpl.class.getDeclaredField("adminV2Topics"); @@ -2917,14 +2861,12 @@ 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") @@ -2999,12 +2941,10 @@ 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(); @@ -3040,7 +2980,6 @@ 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) @@ -3052,14 +2991,12 @@ public void testGetReadPositionWhenJoining() throws Exception { messageId = (MessageIdImpl) producer.send(("Hello Pulsar - " + i).getBytes()); } - List> consumers = new ArrayList<>(); for (int i = 0; i < 2; i++) { - Consumer consumer = pulsarClient.newConsumer() + pulsarClient.newConsumer() .topic(topic) .subscriptionType(SubscriptionType.Key_Shared) .subscriptionName(subName) .subscribe(); - consumers.add(consumer); } TopicStats stats = admin.topics().getStats(topic); @@ -3070,10 +3007,6 @@ 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 @@ -3086,15 +3019,13 @@ public void testPartitionedTopicMsgDelayedAggregated() throws Exception { admin.topics().createPartitionedTopic(topic, numPartitions); for (int i = 0; i < 2; i++) { - @Cleanup - Consumer consumer = pulsarClient.newConsumer() + pulsarClient.newConsumer() .topic(topic) .subscriptionType(SubscriptionType.Shared) .subscriptionName(subName) .subscribe(); } - @Cleanup Producer producer = pulsarClient.newProducer() .topic(topic) .enableBatching(false) From adb4edc0fc6c6996da7dd57d017e917f47bbb52d Mon Sep 17 00:00:00 2001 From: xiangying <1984997880@qq.com> Date: Sun, 26 Feb 2023 15:26:45 +0800 Subject: [PATCH 9/9] fix --- .../broker/namespace/NamespaceOwnershipListenerTests.java | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceOwnershipListenerTests.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceOwnershipListenerTests.java index 6a57307c7a142..5de86180716c5 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceOwnershipListenerTests.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceOwnershipListenerTests.java @@ -19,16 +19,18 @@ package org.apache.pulsar.broker.namespace; import com.google.common.collect.Sets; +import java.util.concurrent.ExecutionException; import java.util.concurrent.atomic.AtomicInteger; import org.apache.pulsar.broker.service.BrokerTestBase; +import org.apache.pulsar.client.admin.PulsarAdminException; import org.apache.pulsar.client.api.Producer; +import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.common.naming.NamespaceBundle; import org.apache.pulsar.common.naming.NamespaceName; import org.testng.Assert; import org.testng.annotations.AfterMethod; import org.testng.annotations.BeforeMethod; import org.testng.annotations.Test; - import java.util.List; import java.util.UUID; import java.util.concurrent.CountDownLatch; @@ -52,7 +54,7 @@ protected void cleanup() throws Exception { } @Test - public void testNamespaceBundleOwnershipListener() throws PulsarAdminException, InterruptedException, PulsarClientException { + public void testNamespaceBundleOwnershipListener() throws Exception { final CountDownLatch countDownLatch = new CountDownLatch(2); final AtomicBoolean onLoad = new AtomicBoolean(false);