From 265bfe724a168a289ca19ae3f123e278c9530b93 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Mon, 28 Aug 2023 16:44:13 +0800 Subject: [PATCH 1/9] remove bundle-data in local metadata store. --- .../pulsar/broker/resources/NamespaceResources.java | 8 +++++--- .../apache/pulsar/broker/resources/PulsarResources.java | 2 +- 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/NamespaceResources.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/NamespaceResources.java index f70d3f31d6e2f..b5ccc9a5a9077 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/NamespaceResources.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/NamespaceResources.java @@ -49,16 +49,18 @@ public class NamespaceResources extends BaseResources { private final IsolationPolicyResources isolationPolicies; private final PartitionedTopicResources partitionedTopicResources; private final MetadataStore configurationStore; + private final MetadataStore localStore; public 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 configurationStore, int operationTimeoutSec) { + public NamespaceResources(MetadataStore localStore, MetadataStore configurationStore, int operationTimeoutSec) { super(configurationStore, Policies.class, operationTimeoutSec); this.configurationStore = configurationStore; isolationPolicies = new IsolationPolicyResources(configurationStore, operationTimeoutSec); partitionedTopicResources = new PartitionedTopicResources(configurationStore, operationTimeoutSec); + this.localStore = localStore; } public CompletableFuture> listNamespacesAsync(String tenant) { @@ -381,13 +383,13 @@ public CompletableFuture runWithMarkDeleteAsync(TopicName topic, // clear resource of `/loadbalance/bundle-data/{tenant}/{namespace}/` in metadata-store public CompletableFuture deleteBundleDataAsync(NamespaceName ns) { final String namespaceBundlePath = joinPath(BUNDLE_DATA_BASE_PATH, ns.toString()); - return getStore().deleteRecursive(namespaceBundlePath); + return this.localStore.deleteRecursive(namespaceBundlePath); } // clear resource of `/loadbalance/bundle-data/{tenant}/` in metadata-store public CompletableFuture deleteBundleDataTenantAsync(String tenant) { final String tenantBundlePath = joinPath(BUNDLE_DATA_BASE_PATH, tenant); - return getStore().deleteRecursive(tenantBundlePath); + return this.localStore.deleteRecursive(tenantBundlePath); } } \ No newline at end of file diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/PulsarResources.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/PulsarResources.java index dfcd0a4194ff5..00f201c5aa1f8 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/PulsarResources.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/PulsarResources.java @@ -60,7 +60,7 @@ public PulsarResources(MetadataStore localMetadataStore, MetadataStore configura if (configurationMetadataStore != null) { tenantResources = new TenantResources(configurationMetadataStore, operationTimeoutSec); clusterResources = new ClusterResources(configurationMetadataStore, operationTimeoutSec); - namespaceResources = new NamespaceResources(configurationMetadataStore, operationTimeoutSec); + namespaceResources = new NamespaceResources(localMetadataStore, configurationMetadataStore, operationTimeoutSec); resourcegroupResources = new ResourceGroupResources(configurationMetadataStore, operationTimeoutSec); } else { tenantResources = null; From 76dc273cc4b4a222b7b174e8e2603d9e39e42653 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Mon, 28 Aug 2023 17:19:50 +0800 Subject: [PATCH 2/9] fix checkstyle. --- .../org/apache/pulsar/broker/resources/PulsarResources.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/PulsarResources.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/PulsarResources.java index 00f201c5aa1f8..a3c5633a6dbe8 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/PulsarResources.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/PulsarResources.java @@ -60,7 +60,8 @@ public PulsarResources(MetadataStore localMetadataStore, MetadataStore configura if (configurationMetadataStore != null) { tenantResources = new TenantResources(configurationMetadataStore, operationTimeoutSec); clusterResources = new ClusterResources(configurationMetadataStore, operationTimeoutSec); - namespaceResources = new NamespaceResources(localMetadataStore, configurationMetadataStore, operationTimeoutSec); + namespaceResources = new NamespaceResources(localMetadataStore, configurationMetadataStore + , operationTimeoutSec); resourcegroupResources = new ResourceGroupResources(configurationMetadataStore, operationTimeoutSec); } else { tenantResources = null; From 73e86e57d0403db480d47df6a54908fed2121f11 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Mon, 28 Aug 2023 18:15:23 +0800 Subject: [PATCH 3/9] add test code. --- .../resources/NamespaceResourcesTest.java | 44 +++++++++++++++++++ 1 file changed, 44 insertions(+) diff --git a/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/resources/NamespaceResourcesTest.java b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/resources/NamespaceResourcesTest.java index 85f54a76dc3c4..deb86e1802f6f 100644 --- a/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/resources/NamespaceResourcesTest.java +++ b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/resources/NamespaceResourcesTest.java @@ -18,12 +18,34 @@ */ package org.apache.pulsar.broker.resources; +import static org.apache.pulsar.broker.resources.BaseResources.joinPath; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertThrows; import static org.testng.Assert.assertTrue; + +import org.apache.pulsar.common.naming.NamespaceName; +import org.apache.pulsar.metadata.api.MetadataStore; +import org.testng.annotations.BeforeMethod; import org.testng.annotations.Test; public class NamespaceResourcesTest { + + private MetadataStore localStore; + private MetadataStore configurationStore; + private NamespaceResources namespaceResources; + + private static final String BUNDLE_DATA_BASE_PATH = "/loadbalance/bundle-data"; + + @BeforeMethod + public void setup() { + localStore = mock(MetadataStore.class); + configurationStore = mock(MetadataStore.class); + namespaceResources = new NamespaceResources(localStore, configurationStore, 30); + } + @Test public void test_pathIsFromNamespace() { assertFalse(NamespaceResources.pathIsFromNamespace("/admin/clusters")); @@ -31,4 +53,26 @@ public void test_pathIsFromNamespace() { assertFalse(NamespaceResources.pathIsFromNamespace("/admin/policies/my-tenant")); assertTrue(NamespaceResources.pathIsFromNamespace("/admin/policies/my-tenant/my-ns")); } + + /** + * Test that the bundle-data node is deleted from the local stores. + */ + @Test + public void testDeleteBundleDataAsync() { + NamespaceName ns = NamespaceName.get("my-tenant/my-ns"); + String namespaceBundlePath = joinPath(BUNDLE_DATA_BASE_PATH, ns.toString()); + namespaceResources.deleteBundleDataAsync(ns); + + String tenant="my-tenant"; + String tenantBundlePath = joinPath(BUNDLE_DATA_BASE_PATH, tenant); + namespaceResources.deleteBundleDataTenantAsync(tenant); + + verify(localStore).deleteRecursive(namespaceBundlePath); + verify(localStore).deleteRecursive(tenantBundlePath); + + assertThrows(()-> verify(configurationStore).deleteRecursive(namespaceBundlePath)); + assertThrows(()-> verify(configurationStore).deleteRecursive(tenantBundlePath)); + } + + } \ No newline at end of file From 23e42c565c60e0dc58681617c869cae12bc5aa84 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Mon, 28 Aug 2023 19:57:35 +0800 Subject: [PATCH 4/9] fix test code. --- .../apache/pulsar/broker/testcontext/PulsarTestContext.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/PulsarTestContext.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/PulsarTestContext.java index 40a42286fda74..aa1ff56c0681a 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/PulsarTestContext.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/PulsarTestContext.java @@ -715,7 +715,8 @@ protected void initializePulsarServices(SpyConfig spyConfig, Builder builder) { if (metadataStore == null) { metadataStore = builder.configurationMetadataStore; } - NamespaceResources nsr = spyConfigPulsarResources.spy(NamespaceResources.class, metadataStore, 30); + NamespaceResources nsr = spyConfigPulsarResources.spy(NamespaceResources.class, + builder.localMetadataStore, metadataStore, 30); TopicResources tsr = spyConfigPulsarResources.spy(TopicResources.class, metadataStore); pulsarResources( spyConfigPulsarResources.spy( From e6561391dc2d7dcc4421af19a063b811304aad19 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Tue, 29 Aug 2023 10:35:30 +0800 Subject: [PATCH 5/9] add test code. --- .../test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java | 2 ++ 1 file changed, 2 insertions(+) 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 0adf945a55552..6c8889b5f8dac 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 @@ -1720,6 +1720,8 @@ public void testDeleteNamespace(NamespaceAttr namespaceAttr) throws Exception { // Set conf. cleanup(); setNamespaceAttr(namespaceAttr); + this.conf.setMetadataStoreUrl("127.0.0.1:2181"); + this.conf.setConfigurationMetadataStoreUrl("127.0.0.1:2182"); setup(); String tenant = newUniqueName("test-tenant"); From 8e4473d41d5a79e7aa434060e4b5888dba78c9ff Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Tue, 29 Aug 2023 15:15:44 +0800 Subject: [PATCH 6/9] add test code. --- .../pulsar/broker/admin/AdminApi2Test.java | 28 +++++++++++++++---- 1 file changed, 23 insertions(+), 5 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 6c8889b5f8dac..af8ac93adb780 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 @@ -125,6 +125,8 @@ import org.apache.pulsar.common.policies.data.TopicStats; import org.apache.pulsar.common.policies.data.TopicType; import org.apache.pulsar.common.policies.data.impl.BacklogQuotaImpl; +import org.apache.pulsar.metadata.api.MetadataCache; +import org.apache.pulsar.policies.data.loadbalancer.BundleData; import org.awaitility.Awaitility; import org.testng.Assert; import org.testng.annotations.AfterClass; @@ -1713,6 +1715,12 @@ private void setNamespaceAttr(NamespaceAttr namespaceAttr){ conf.setForceDeleteNamespaceAllowed(namespaceAttr.forceDeleteNamespaceAllowed); } + private static Object getField(Object obj, String fieldName) throws Exception { + Field field = obj.getClass().getDeclaredField(fieldName); + field.setAccessible(true); + return field.get(obj); + } + @Test(dataProvider = "namespaceAttrs") public void testDeleteNamespace(NamespaceAttr namespaceAttr) throws Exception { restartClusterAfterTest(); @@ -1742,6 +1750,20 @@ public void testDeleteNamespace(NamespaceAttr namespaceAttr) throws Exception { admin.topics().createPartitionedTopic(topic, 10); assertFalse(admin.topics().getList(namespace).isEmpty()); + // export bundle-data to metadata store + final String managedLedgersPath = "/managed-ledgers/" + namespace; + final String bundleDataPath = "/loadbalance/bundle-data/" + namespace; + ModularLoadManagerImpl loadManager = (ModularLoadManagerImpl) getField( + pulsar.getLoadManager().get(), "loadManager"); + MetadataCache bundlesCache = (MetadataCache) getField( + loadManager, "bundlesCache"); + final BundleData bundleData = new BundleData(10, 1000); + bundlesCache.readModifyUpdateOrCreate(bundleDataPath, __ -> bundleData).join(); + + // assert znode exists in metadata store + assertTrue(pulsar.getLocalMetadataStore().exists(bundleDataPath).join()); + assertTrue(pulsar.getLocalMetadataStore().exists(managedLedgersPath).join()); + try { admin.namespaces().deleteNamespace(namespace, false); fail("should have failed due to namespace not empty"); @@ -1758,12 +1780,8 @@ public void testDeleteNamespace(NamespaceAttr namespaceAttr) throws Exception { assertFalse(admin.namespaces().getNamespaces(tenant).contains(namespace)); assertTrue(admin.namespaces().getNamespaces(tenant).isEmpty()); - - final String managedLedgersPath = "/managed-ledgers/" + namespace; + // assert znode deleted in metadata store assertFalse(pulsar.getLocalMetadataStore().exists(managedLedgersPath).join()); - - - final String bundleDataPath = "/loadbalance/bundle-data/" + namespace; assertFalse(pulsar.getLocalMetadataStore().exists(bundleDataPath).join()); } From a9472bc64bfbb6be3b4aac7654bd76b9f7fdf0b3 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Wed, 30 Aug 2023 16:24:59 +0800 Subject: [PATCH 7/9] add test code. --- .../pulsar/broker/admin/AdminApi2Test.java | 23 +++++++++++++------ 1 file changed, 16 insertions(+), 7 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 af8ac93adb780..bf27d8a499f8d 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 @@ -67,6 +67,7 @@ import org.apache.pulsar.broker.admin.AdminApiTest.MockedPulsarService; import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest; import org.apache.pulsar.broker.loadbalance.impl.ModularLoadManagerImpl; +import org.apache.pulsar.broker.loadbalance.impl.ModularLoadManagerWrapper; import org.apache.pulsar.broker.loadbalance.impl.SimpleLoadManagerImpl; import org.apache.pulsar.broker.service.Topic; import org.apache.pulsar.broker.service.persistent.PersistentSubscription; @@ -1750,15 +1751,23 @@ public void testDeleteNamespace(NamespaceAttr namespaceAttr) throws Exception { admin.topics().createPartitionedTopic(topic, 10); assertFalse(admin.topics().getList(namespace).isEmpty()); - // export bundle-data to metadata store final String managedLedgersPath = "/managed-ledgers/" + namespace; final String bundleDataPath = "/loadbalance/bundle-data/" + namespace; - ModularLoadManagerImpl loadManager = (ModularLoadManagerImpl) getField( - pulsar.getLoadManager().get(), "loadManager"); - MetadataCache bundlesCache = (MetadataCache) getField( - loadManager, "bundlesCache"); - final BundleData bundleData = new BundleData(10, 1000); - bundlesCache.readModifyUpdateOrCreate(bundleDataPath, __ -> bundleData).join(); + // Trigger bundle owned by brokers. + pulsarClient.newProducer().topic(topic).create().close(); + // Trigger bundle data write to ZK. + Awaitility.await().untilAsserted(() -> { + boolean bundleDataWereWriten = false; + for (PulsarService ps : new PulsarService[]{pulsar, mockPulsarSetup.getPulsar()}) { + ModularLoadManagerWrapper loadManager = (ModularLoadManagerWrapper) ps.getLoadManager().get(); + ModularLoadManagerImpl loadManagerImpl = (ModularLoadManagerImpl) loadManager.getLoadManager(); + ps.getBrokerService().updateRates(); + loadManagerImpl.updateLocalBrokerData(); + loadManagerImpl.writeBundleDataOnZooKeeper(); + bundleDataWereWriten = bundleDataWereWriten || ps.getLocalMetadataStore().exists(bundleDataPath).join(); + } + assertTrue(bundleDataWereWriten); + }); // assert znode exists in metadata store assertTrue(pulsar.getLocalMetadataStore().exists(bundleDataPath).join()); From efb87b1e5a366d1d074da929608296aee084f1ad Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Wed, 30 Aug 2023 16:57:40 +0800 Subject: [PATCH 8/9] fix checkstyle. --- .../test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java | 2 -- 1 file changed, 2 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 bf27d8a499f8d..d15272c15a19c 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 @@ -126,8 +126,6 @@ import org.apache.pulsar.common.policies.data.TopicStats; import org.apache.pulsar.common.policies.data.TopicType; import org.apache.pulsar.common.policies.data.impl.BacklogQuotaImpl; -import org.apache.pulsar.metadata.api.MetadataCache; -import org.apache.pulsar.policies.data.loadbalancer.BundleData; import org.awaitility.Awaitility; import org.testng.Assert; import org.testng.annotations.AfterClass; From cc9bf78881a10b74284b46b9cf64f2398be56f3c Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Wed, 30 Aug 2023 17:33:51 +0800 Subject: [PATCH 9/9] fix. --- .../java/org/apache/pulsar/broker/admin/AdminApi2Test.java | 6 ------ 1 file changed, 6 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 d15272c15a19c..99009a1988206 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 @@ -1714,12 +1714,6 @@ private void setNamespaceAttr(NamespaceAttr namespaceAttr){ conf.setForceDeleteNamespaceAllowed(namespaceAttr.forceDeleteNamespaceAllowed); } - private static Object getField(Object obj, String fieldName) throws Exception { - Field field = obj.getClass().getDeclaredField(fieldName); - field.setAccessible(true); - return field.get(obj); - } - @Test(dataProvider = "namespaceAttrs") public void testDeleteNamespace(NamespaceAttr namespaceAttr) throws Exception { restartClusterAfterTest();