You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by xi...@apache.org on 2023/02/25 08:25:25 UTC
[pulsar] 02/02: [fix][test] Fix flaky test deleteNamespaceGracefully (#18220)
This is an automated email from the ASF dual-hosted git repository.
xiangying pushed a commit to branch xaingying/cherry-pick-2.10/improve_test
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit a3e593abbcb3a88509816ccf584823478f21aa74
Author: fengyubiao <99...@qq.com>
AuthorDate: Thu Nov 24 15:20:44 2022 +0800
[fix][test] Fix flaky test deleteNamespaceGracefully (#18220)
(cherry picked from commit c544ea3aae1a8be8e7bcd4f89fb1899ce0a3350f)
---
.../apache/pulsar/broker/admin/AdminApi2Test.java | 8 +-
.../apache/pulsar/broker/admin/AdminApiTest.java | 38 +++---
.../pulsar/broker/admin/AdminApiTlsAuthTest.java | 2 +-
.../broker/auth/MockedPulsarServiceBaseTest.java | 142 +++++++++++++++++++++
.../namespace/NamespaceOwnershipListenerTests.java | 13 +-
.../OwnerShipForCurrentServerTestBase.java | 3 +-
.../ResourceGroupConfigListenerTest.java | 2 +-
.../broker/service/BacklogQuotaManagerTest.java | 25 +++-
.../BrokerServiceBundlesCacheInvalidationTest.java | 2 +-
.../broker/service/PersistentTopicE2ETest.java | 4 +-
.../broker/transaction/TransactionTestBase.java | 21 ++-
.../client/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 fcc2f144907..3c99452fdc0 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 class AdminApi2Test extends MockedPulsarServiceBaseTest {
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 class AdminApi2Test extends MockedPulsarServiceBaseTest {
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 class AdminApi2Test extends MockedPulsarServiceBaseTest {
});
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 class AdminApi2Test extends MockedPulsarServiceBaseTest {
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 3f151daf8d2..a9ba361bbb9 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 class AdminApiTest extends MockedPulsarServiceBaseTest {
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 class AdminApiTest extends MockedPulsarServiceBaseTest {
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 class AdminApiTest extends MockedPulsarServiceBaseTest {
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 class AdminApiTest extends MockedPulsarServiceBaseTest {
}
@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 class AdminApiTest extends MockedPulsarServiceBaseTest {
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 class AdminApiTest extends MockedPulsarServiceBaseTest {
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 class AdminApiTest extends MockedPulsarServiceBaseTest {
}
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 class AdminApiTest extends MockedPulsarServiceBaseTest {
@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 class AdminApiTest extends MockedPulsarServiceBaseTest {
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 class AdminApiTest extends MockedPulsarServiceBaseTest {
}
// 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 class AdminApiTest extends MockedPulsarServiceBaseTest {
@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 class AdminApiTest extends MockedPulsarServiceBaseTest {
@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 class AdminApiTest extends MockedPulsarServiceBaseTest {
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 class AdminApiTest extends MockedPulsarServiceBaseTest {
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 class AdminApiTest extends MockedPulsarServiceBaseTest {
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 311e31be735..11b806a656f 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 class AdminApiTlsAuthTest extends MockedPulsarServiceBaseTest {
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 84964680d34..4f3af27a384 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.URI;
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.PulsarService;
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.pulsar.tests.TestRetrySupport;
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 @@ public abstract class MockedPulsarServiceBaseTest extends TestRetrySupport {
}
}
+ protected Object asyncRequests(Consumer<TestAsyncResponse> 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<Class<?>> register(Class<?> callback) {
+ return null;
+ }
+
+ @Override
+ public Map<Class<?>, Collection<Class<?>>> register(Class<?> callback, Class<?>... callbacks) {
+ return null;
+ }
+
+ @Override
+ public Collection<Class<?>> register(Object callback) {
+ return null;
+ }
+
+ @Override
+ public Map<Class<?>, Collection<Class<?>>> 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<PulsarService> 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 c0a46881628..2738ab897d2 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 @@ public class NamespaceOwnershipListenerTests extends BrokerTestBase {
}
@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 class NamespaceOwnershipListenerTests extends BrokerTestBase {
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 class NamespaceOwnershipListenerTests extends BrokerTestBase {
}
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 class NamespaceOwnershipListenerTests extends BrokerTestBase {
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 c25c6c23fb0..80558361f4b 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 @@ public class OwnerShipForCurrentServerTestBase {
MockZooKeeperSession mockZooKeeperSession = MockZooKeeperSession.newInstance(mockZooKeeper);
doReturn(new ZKMetadataStore(mockZooKeeperSession)).when(pulsar).createLocalMetadataStore();
doReturn(new ZKMetadataStore(mockZooKeeperSession)).when(pulsar).createConfigurationMetadataStore();
- Supplier<NamespaceService> namespaceServiceSupplier = () -> spyWithClassAndConstructorArgs(NamespaceService.class, pulsar);
+ Supplier<NamespaceService> 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 38cae3253fe..64c5fee14d0 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 class ResourceGroupConfigListenerTest extends MockedPulsarServiceBaseTest
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 1d2878ca644..96fc452745b 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.Position;
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 @@ public class BacklogQuotaManagerTest {
}
@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 573d473ed58..a55540c3f4f 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 class BrokerServiceBundlesCacheInvalidationTest extends BrokerTestBase {
// 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 54dc706016b..fdd1170c21b 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 class PersistentTopicE2ETest extends BrokerTestBase {
consumer.close();
admin.topics().deleteSubscription(topicName, subName);
admin.topics().delete(topicName);
- admin.namespaces().deleteNamespace(namespaceName);
+ deleteNamespaceWithRetry(namespaceName, false);
}
@Test
@@ -1052,7 +1052,7 @@ public class PersistentTopicE2ETest extends BrokerTestBase {
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 25c555f09b9..990f63011b3 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.bookkeeper.util.ZkUtils;
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 @@ public abstract class TransactionTestBase extends TestRetrySupport {
MockZooKeeperSession mockZooKeeperSession = MockZooKeeperSession.newInstance(mockZooKeeper);
doReturn(new ZKMetadataStore(mockZooKeeperSession)).when(pulsar).createLocalMetadataStore();
doReturn(new ZKMetadataStore(mockZooKeeperSession)).when(pulsar).createConfigurationMetadataStore();
- Supplier<NamespaceService> namespaceServiceSupplier = () -> spyWithClassAndConstructorArgs(NamespaceService.class, pulsar);
+ Supplier<NamespaceService> namespaceServiceSupplier =
+ () -> spyWithClassAndConstructorArgs(NamespaceService.class, pulsar);
doReturn(namespaceServiceSupplier).when(pulsar).getNamespaceServiceProvider();
SameThreadOrderedSafeExecutor executor = new SameThreadOrderedSafeExecutor();
@@ -330,4 +333,20 @@ public abstract class TransactionTestBase extends TestRetrySupport {
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 1a56d689799..df695cbf715 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 class DispatcherBlockConsumerTest extends ProducerConsumerBase {
// 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<NamespaceService> namespaceServiceSupplier = () -> spyWithClassAndConstructorArgs(NamespaceService.class, pulsar);
+ Supplier<NamespaceService> namespaceServiceSupplier =
+ () -> spyWithClassAndConstructorArgs(NamespaceService.class, pulsar);
doReturn(namespaceServiceSupplier).when(pulsar).getNamespaceServiceProvider();
}
stopBroker();