fix:remove the loadbalance/bundle-data node (apache#13164)

Fixes apache#13162 ### Motivation fix: remove the loadbalance/bundle-data node, when deleting namespace. it will cause zk node leak, fix apache#13162 ### Modifications remove the bundle-data Recursively
bestming8 · Dec 8, 2021 · 1220f84 · 1220f84
1 parent a89c6e4
commit 1220f84
Show file tree

Hide file tree

Showing 4 changed files with 46 additions and 0 deletions.
diff --git a/...ar-broker-common/src/main/java/org/apache/pulsar/broker/resources/NamespaceResources.java b/...ar-broker-common/src/main/java/org/apache/pulsar/broker/resources/NamespaceResources.java
@@ -54,6 +54,7 @@ public class NamespaceResources extends BaseResources<Policies> {
 
     private static final String POLICIES_READONLY_FLAG_PATH = "/admin/flags/policies-readonly";
     private static final String NAMESPACE_BASE_PATH = "/namespace";
+    private static final String BUNDLE_DATA_BASE_PATH = "/loadbalance/bundle-data";
 
     public NamespaceResources(MetadataStore localStore, MetadataStore configurationStore, int operationTimeoutSec) {
         super(configurationStore, Policies.class, operationTimeoutSec);
@@ -296,4 +297,39 @@ public CompletableFuture<Void> clearPartitionedTopicTenantAsync(String tenant) {
             return future;
         }
     }
+
+    // clear resource of `/loadbalance/bundle-data/{tenant}/{namespace}/` for zk-node
+    public CompletableFuture<Void> deleteBundleDataAsync(NamespaceName ns) {
+        final String namespaceBundlePath = joinPath(BUNDLE_DATA_BASE_PATH, ns.toString());
+        CompletableFuture<Void> future = new CompletableFuture<Void>();
+        deleteRecursiveAsync(this, namespaceBundlePath).whenComplete((ignore, ex) -> {
+            if (ex instanceof MetadataStoreException.NotFoundException) {
+                future.complete(null);
+            } else if (ex != null) {
+                future.completeExceptionally(ex);
+            } else {
+                future.complete(null);
+            }
+        });
+
+        return future;
+    }
+
+    // clear resource of `/loadbalance/bundle-data/{tenant}/` for zk-node
+    public CompletableFuture<Void> deleteBundleDataTenantAsync(String tenant) {
+        final String tenantBundlePath = joinPath(BUNDLE_DATA_BASE_PATH, tenant);
+        CompletableFuture<Void> future = new CompletableFuture<Void>();
+        deleteRecursiveAsync(this, tenantBundlePath).whenComplete((ignore, ex) -> {
+            if (ex instanceof MetadataStoreException.NotFoundException) {
+                future.complete(null);
+            } else if (ex != null) {
+                future.completeExceptionally(ex);
+            } else {
+                future.complete(null);
+            }
+        });
+
+        return future;
+    }
+
 }
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
@@ -320,6 +320,8 @@ protected void internalClearZkSources(AsyncResponse asyncResponse) {
                 .thenCompose(ignore -> namespaceResources().deletePoliciesAsync(namespaceName))
                 // clear z-node of local policies
                 .thenCompose(ignore -> getLocalPolicies().deleteLocalPoliciesAsync(namespaceName))
+                // clear /loadbalance/bundle-data
+                .thenCompose(ignore -> namespaceResources().deleteBundleDataAsync(namespaceName))
                 .whenComplete((ignore, ex) -> {
                     if (ex != null) {
                         log.warn("[{}] Failed to remove namespace or managed-ledger for {}",

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/TenantsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/TenantsBase.java
@@ -269,6 +269,8 @@ protected void internalDeleteTenant(AsyncResponse asyncResponse, String tenant)
                             .getPartitionedTopicResources().clearPartitionedTopicTenantAsync(tenant))
                     .thenCompose(ignore -> pulsar().getPulsarResources().getLocalPolicies()
                             .deleteLocalPoliciesTenantAsync(tenant))
+                    .thenCompose(ignore -> pulsar().getPulsarResources().getNamespaceResources()
+                            .deleteBundleDataTenantAsync(tenant))
                     .whenComplete((ignore, ex) -> {
                         if (ex != null) {
                             log.error("[{}] Failed to delete tenant {}", clientAppId(), tenant, ex);

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest2.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest2.java
@@ -1311,9 +1311,11 @@ public void testDeleteTenant() throws Exception {
         final String managedLedgersPath = "/managed-ledgers/" + tenant;
         final String partitionedTopicPath = "/admin/partitioned-topics/" + tenant;
         final String localPoliciesPath = "/admin/local-policies/" + tenant;
+        final String bundleDataPath = "/loadbalance/bundle-data/" + tenant;
         assertFalse(pulsar.getLocalMetadataStore().exists(managedLedgersPath).join());
         assertFalse(pulsar.getLocalMetadataStore().exists(partitionedTopicPath).join());
         assertFalse(pulsar.getLocalMetadataStore().exists(localPoliciesPath).join());
+        assertFalse(pulsar.getLocalMetadataStore().exists(bundleDataPath).join());
     }
 
     @Test
@@ -1357,6 +1359,10 @@ public void testDeleteNamespace() throws Exception {
 
         final String managedLedgersPath = "/managed-ledgers/" + namespace;
         assertFalse(pulsar.getLocalMetadataStore().exists(managedLedgersPath).join());
+
+
+        final String bundleDataPath = "/loadbalance/bundle-data/" + namespace;
+        assertFalse(pulsar.getLocalMetadataStore().exists(bundleDataPath).join());
     }