[fix][admin] Fix can't delete tenant for v1 (apache#22550)

(cherry picked from commit 5d0f1d6)
datastax · Jun 4, 2024 · ce82219 · ce82219
1 parent 24432eb
commit ce82219
Show file tree

Hide file tree

Showing 2 changed files with 30 additions and 1 deletion.
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/TopicResources.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/TopicResources.java
@@ -101,7 +101,7 @@ public CompletableFuture<Void> clearDomainPersistence(NamespaceName ns) {
     public CompletableFuture<Void> clearTenantPersistence(String tenant) {
         String path = MANAGED_LEDGER_PATH + "/" + tenant;
         log.info("Clearing tenant persistence for tenant: {}, path {}", tenant, path);
-        return store.deleteIfExists(path, Optional.empty());
+        return store.deleteRecursive(path);
     }
 
     void handleNotification(Notification notification) {

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/AuthorizationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/AuthorizationTest.java
@@ -32,6 +32,7 @@
 import org.apache.pulsar.broker.resources.PulsarResources;
 import org.apache.pulsar.client.admin.PulsarAdmin;
 import org.apache.pulsar.client.admin.PulsarAdminBuilder;
+import org.apache.pulsar.client.api.ClientBuilder;
 import org.apache.pulsar.common.naming.TopicDomain;
 import org.apache.pulsar.common.naming.TopicName;
 import org.apache.pulsar.common.policies.data.AuthAction;
@@ -55,12 +56,17 @@ public AuthorizationTest() {
     @Override
     public void setup() throws Exception {
         conf.setClusterName("c1");
+        conf.setSystemTopicEnabled(false);
         conf.setAuthenticationEnabled(true);
+        conf.setForceDeleteNamespaceAllowed(true);
+        conf.setForceDeleteTenantAllowed(true);
         conf.setAuthenticationProviders(
                 Sets.newHashSet("org.apache.pulsar.broker.auth.MockAuthenticationProvider"));
         conf.setAuthorizationEnabled(true);
         conf.setAuthorizationAllowWildcardsMatching(true);
         conf.setSuperUserRoles(Sets.newHashSet("pulsar.super_user", "pass.pass"));
+        conf.setBrokerClientAuthenticationPlugin(MockAuthentication.class.getName());
+        conf.setBrokerClientAuthenticationParameters("user:pass.pass");
         internalSetup();
     }
 
@@ -69,6 +75,11 @@ protected void customizeNewPulsarAdminBuilder(PulsarAdminBuilder pulsarAdminBuil
         pulsarAdminBuilder.authentication(new MockAuthentication("pass.pass"));
     }
 
+    @Override
+    protected void customizeNewPulsarClientBuilder(ClientBuilder clientBuilder) {
+        clientBuilder.authentication(new MockAuthentication("pass.pass"));
+    }
+
     @AfterClass(alwaysRun = true)
     @Override
     public void cleanup() throws Exception {
@@ -232,6 +243,24 @@ public void simple() throws Exception {
 
         admin.namespaces().deleteNamespace("p1/c1/ns1");
         admin.tenants().deleteTenant("p1");
+
+        admin.clusters().deleteCluster("c1");
+    }
+
+    @Test
+    public void testDeleteV1Tenant() throws Exception {
+        admin.clusters().createCluster("c1", ClusterData.builder().build());
+        admin.tenants().createTenant("p1", new TenantInfoImpl(Sets.newHashSet("role1"), Sets.newHashSet("c1")));
+        waitForChange();
+        admin.namespaces().createNamespace("p1/c1/ns1");
+        waitForChange();
+
+
+        String topic = "persistent://p1/c1/ns1/ds2";
+        admin.topics().createNonPartitionedTopic(topic);
+
+        admin.namespaces().deleteNamespace("p1/c1/ns1", true);
+        admin.tenants().deleteTenant("p1", true);
         admin.clusters().deleteCluster("c1");
     }