From 46e22afe26cc47ab963f6fb13c9a09ed8c149ca0 Mon Sep 17 00:00:00 2001 From: houxiaoyu Date: Mon, 4 Sep 2023 16:54:07 +0800 Subject: [PATCH 01/10] Add BundleDataResources --- .../broker/resources/BundleDataResources.java | 64 +++++++++++++++++++ .../broker/resources/NamespaceResources.java | 17 +---- .../broker/resources/PulsarResources.java | 7 +- .../resources/BundleDataResourcesTest.java | 44 +++++++++++++ .../resources/NamespaceResourcesTest.java | 42 ------------ .../broker/admin/impl/NamespacesBase.java | 2 +- .../pulsar/broker/admin/impl/TenantsBase.java | 2 +- .../impl/ModularLoadManagerImpl.java | 17 +---- .../pulsar/broker/web/PulsarWebResource.java | 5 ++ .../impl/ModularLoadManagerImplTest.java | 7 +- 10 files changed, 128 insertions(+), 79 deletions(-) create mode 100644 pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/BundleDataResources.java create mode 100644 pulsar-broker-common/src/test/java/org/apache/pulsar/broker/resources/BundleDataResourcesTest.java diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/BundleDataResources.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/BundleDataResources.java new file mode 100644 index 0000000000000..e99f744896c4a --- /dev/null +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/BundleDataResources.java @@ -0,0 +1,64 @@ +/* + * 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.broker.resources; + +import com.google.common.annotations.VisibleForTesting; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; +import org.apache.pulsar.common.naming.NamespaceName; +import org.apache.pulsar.metadata.api.MetadataStore; +import org.apache.pulsar.policies.data.loadbalancer.BundleData; + +public class BundleDataResources extends BaseResources { + @VisibleForTesting + public static final String BUNDLE_DATA_BASE_PATH = "/loadbalance/bundle-data"; + + public BundleDataResources(MetadataStore store, int operationTimeoutSec) { + super(store, BundleData.class, operationTimeoutSec); + } + + public CompletableFuture> getBundleData(String bundle) { + return getAsync(getBundleDataPath(bundle)); + } + + public CompletableFuture updateBundleData(String bundle, BundleData data) { + return setWithCreateAsync(getBundleDataPath(bundle), __ -> data); + } + + public CompletableFuture deleteBundleData(String bundle) { + return deleteAsync(getBundleDataPath(bundle)); + } + + // 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); + } + + // 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); + } + + // Get the metadata store path for the given bundle full name. + private String getBundleDataPath(final String bundle) { + return BUNDLE_DATA_BASE_PATH + "/" + bundle; + } +} 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 b5ccc9a5a9077..982171cf3f92b 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,18 +49,16 @@ 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 localStore, MetadataStore configurationStore, int operationTimeoutSec) { + public NamespaceResources(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) { @@ -379,17 +377,4 @@ public CompletableFuture runWithMarkDeleteAsync(TopicName topic, return future; } } - - // 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 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 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 a3c5633a6dbe8..ba32ebc3b973f 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 @@ -48,6 +48,8 @@ public class PulsarResources { @Getter private final TopicResources topicResources; @Getter + private final BundleDataResources bundleDataResources; + @Getter private final Optional localMetadataStore; @Getter private final Optional configurationMetadataStore; @@ -60,8 +62,7 @@ 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, operationTimeoutSec); resourcegroupResources = new ResourceGroupResources(configurationMetadataStore, operationTimeoutSec); } else { tenantResources = null; @@ -76,12 +77,14 @@ public PulsarResources(MetadataStore localMetadataStore, MetadataStore configura loadReportResources = new LoadManagerReportResources(localMetadataStore, operationTimeoutSec); bookieResources = new BookieResources(localMetadataStore, operationTimeoutSec); topicResources = new TopicResources(localMetadataStore); + bundleDataResources = new BundleDataResources(localMetadataStore, operationTimeoutSec); } else { dynamicConfigResources = null; localPolicies = null; loadReportResources = null; bookieResources = null; topicResources = null; + bundleDataResources = null; } this.localMetadataStore = Optional.ofNullable(localMetadataStore); diff --git a/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/resources/BundleDataResourcesTest.java b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/resources/BundleDataResourcesTest.java new file mode 100644 index 0000000000000..64f9f7733e2c3 --- /dev/null +++ b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/resources/BundleDataResourcesTest.java @@ -0,0 +1,44 @@ +package org.apache.pulsar.broker.resources; + +import static org.apache.pulsar.broker.resources.BaseResources.joinPath; +import static org.apache.pulsar.broker.resources.BundleDataResources.BUNDLE_DATA_BASE_PATH; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.testng.Assert.assertThrows; +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 BundleDataResourcesTest { + private MetadataStore configurationStore; + private MetadataStore localStore; + private BundleDataResources bundleDataResources; + + @BeforeMethod + public void setup() { + localStore = mock(MetadataStore.class); + configurationStore = mock(MetadataStore.class); + bundleDataResources = new BundleDataResources(localStore, 30); + } + + /** + * 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()); + bundleDataResources.deleteBundleDataAsync(ns); + + String tenant="my-tenant"; + String tenantBundlePath = joinPath(BUNDLE_DATA_BASE_PATH, tenant); + bundleDataResources.deleteBundleDataTenantAsync(tenant); + + verify(localStore).deleteRecursive(namespaceBundlePath); + verify(localStore).deleteRecursive(tenantBundlePath); + + assertThrows(()-> verify(configurationStore).deleteRecursive(namespaceBundlePath)); + assertThrows(()-> verify(configurationStore).deleteRecursive(tenantBundlePath)); + } +} 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 deb86e1802f6f..7fb9e2c476d08 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,34 +18,12 @@ */ 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")); @@ -54,25 +32,5 @@ public void test_pathIsFromNamespace() { 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 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 index 4c364068077d9..f7f0d18735f0c 100644 --- 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 @@ -468,7 +468,7 @@ protected CompletableFuture internalClearZkSources() { // clear z-node of local policies .thenCompose(ignore -> getLocalPolicies().deleteLocalPoliciesAsync(namespaceName)) // clear /loadbalance/bundle-data - .thenCompose(ignore -> namespaceResources().deleteBundleDataAsync(namespaceName)); + .thenCompose(ignore -> bundleDataResources().deleteBundleDataAsync(namespaceName)); } 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 index b93f3e3c6ebcc..6f18c55b4e89c 100644 --- 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 @@ -236,7 +236,7 @@ protected CompletableFuture internalDeleteTenantAsync(String tenant) { .getPartitionedTopicResources().clearPartitionedTopicTenantAsync(tenant)) .thenCompose(__ -> pulsar().getPulsarResources().getLocalPolicies() .deleteLocalPoliciesTenantAsync(tenant)) - .thenCompose(__ -> pulsar().getPulsarResources().getNamespaceResources() + .thenCompose(__ -> pulsar().getPulsarResources().getBundleDataResources() .deleteBundleDataTenantAsync(tenant)); } 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 0d5dbf489e90f..9b7b1ae9758fb 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 @@ -91,9 +91,6 @@ public class ModularLoadManagerImpl implements ModularLoadManager { private static final Logger log = LoggerFactory.getLogger(ModularLoadManagerImpl.class); - // Path to ZNode whose children contain BundleData jsons for each bundle (new API version of ResourceQuota). - public static final String BUNDLE_DATA_PATH = "/loadbalance/bundle-data"; - // Default message rate to assume for unseen bundles. public static final double DEFAULT_MESSAGE_RATE = 50; @@ -120,7 +117,6 @@ public class ModularLoadManagerImpl implements ModularLoadManager { private LockManager brokersData; private ResourceLock brokerDataLock; - private MetadataCache bundlesCache; private MetadataCache resourceQuotaCache; private MetadataCache timeAverageBrokerDataCache; @@ -244,7 +240,6 @@ public boolean isEnableNonPersistentTopics(String brokerUrl) { public void initialize(final PulsarService pulsar) { this.pulsar = pulsar; brokersData = pulsar.getCoordinationService().getLockManager(LocalBrokerData.class); - bundlesCache = pulsar.getLocalMetadataStore().getMetadataCache(BundleData.class); resourceQuotaCache = pulsar.getLocalMetadataStore().getMetadataCache(ResourceQuota.class); timeAverageBrokerDataCache = pulsar.getLocalMetadataStore().getMetadataCache(TimeAverageBrokerData.class); pulsar.getLocalMetadataStore().registerListener(this::handleDataNotification); @@ -381,7 +376,7 @@ public CompletableFuture> getAvailableBrokersAsync() { public BundleData getBundleDataOrDefault(final String bundle) { BundleData bundleData = null; try { - Optional optBundleData = bundlesCache.get(getBundleDataPath(bundle)).join(); + Optional optBundleData = pulsar.getPulsarResources().getBundleDataResources().getBundleData(bundle).join(); if (optBundleData.isPresent()) { return optBundleData.get(); } @@ -418,11 +413,6 @@ public BundleData getBundleDataOrDefault(final String bundle) { return bundleData; } - // Get the metadata store path for the given bundle full name. - public static String getBundleDataPath(final String bundle) { - return BUNDLE_DATA_PATH + "/" + bundle; - } - // Use the Pulsar client to acquire the namespace bundle stats. private Map getBundleStats() { return pulsar.getBrokerService().getBundleStats(); @@ -1151,8 +1141,7 @@ public void writeBundleDataOnZooKeeper() { for (Map.Entry entry : loadData.getBundleData().entrySet()) { final String bundle = entry.getKey(); final BundleData data = entry.getValue(); - futures.add(bundlesCache.readModifyUpdateOrCreate(getBundleDataPath(bundle), __ -> data) - .thenApply(__ -> null)); + futures.add(pulsar.getPulsarResources().getBundleDataResources().updateBundleData(bundle, data)); } // Write the time average broker data to metadata store. @@ -1173,7 +1162,7 @@ public void writeBundleDataOnZooKeeper() { private void deleteBundleDataFromMetadataStore(String bundle) { try { - bundlesCache.delete(getBundleDataPath(bundle)).join(); + pulsar.getPulsarResources().getBundleDataResources().deleteBundleData(bundle).join(); } catch (Exception e) { if (!(e.getCause() instanceof NotFoundException)) { log.warn("Failed to delete bundle-data {} from metadata store", bundle, e); 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 e65ef50c72aff..2d7247acc27f4 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 @@ -60,6 +60,7 @@ import org.apache.pulsar.broker.namespace.LookupOptions; import org.apache.pulsar.broker.namespace.NamespaceService; import org.apache.pulsar.broker.resources.BookieResources; +import org.apache.pulsar.broker.resources.BundleDataResources; import org.apache.pulsar.broker.resources.ClusterResources; import org.apache.pulsar.broker.resources.DynamicConfigurationResources; import org.apache.pulsar.broker.resources.LocalPoliciesResources; @@ -1111,6 +1112,10 @@ protected NamespaceResources namespaceResources() { return pulsar().getPulsarResources().getNamespaceResources(); } + protected BundleDataResources bundleDataResources() { + return pulsar().getPulsarResources().getBundleDataResources(); + } + protected ResourceGroupResources resourceGroupResources() { return pulsar().getPulsarResources().getResourcegroupResources(); } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImplTest.java index d8acb6d24e9ef..6bc8196ae6ecc 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImplTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImplTest.java @@ -63,6 +63,7 @@ import org.apache.pulsar.broker.loadbalance.LoadData; import org.apache.pulsar.broker.loadbalance.LoadManager; import org.apache.pulsar.broker.loadbalance.impl.LoadManagerShared.BrokerTopicLoadingPredicate; +import org.apache.pulsar.broker.resources.BundleDataResources; import org.apache.pulsar.client.admin.Namespaces; import org.apache.pulsar.client.admin.PulsarAdmin; import org.apache.pulsar.client.admin.PulsarAdminException; @@ -290,7 +291,7 @@ public void testEvenBundleDistribution() throws Exception { final TimeAverageMessageData longTermMessageData = new TimeAverageMessageData(1000); longTermMessageData.setMsgRateIn(1000); bundleData.setLongTermData(longTermMessageData); - final String firstBundleDataPath = String.format("%s/%s", ModularLoadManagerImpl.BUNDLE_DATA_PATH, bundles[0]); + final String firstBundleDataPath = String.format("%s/%s", BundleDataResources.BUNDLE_DATA_BASE_PATH, bundles[0]); // Write long message rate for first bundle to ensure that even bundle distribution is not a coincidence of // balancing by message rate. If we were balancing by message rate, one of the brokers should only have this // one bundle. @@ -386,7 +387,7 @@ public void testMaxTopicDistributionToBroker() throws Exception { final TimeAverageMessageData longTermMessageData = new TimeAverageMessageData(1000); longTermMessageData.setMsgRateIn(1000); bundleData.setLongTermData(longTermMessageData); - final String firstBundleDataPath = String.format("%s/%s", ModularLoadManagerImpl.BUNDLE_DATA_PATH, bundles[0]); + final String firstBundleDataPath = String.format("%s/%s", BundleDataResources.BUNDLE_DATA_BASE_PATH, bundles[0]); pulsar1.getLocalMetadataStore().getMetadataCache(BundleData.class).create(firstBundleDataPath, bundleData).join(); String maxTopicOwnedBroker = primaryLoadManager.selectBrokerForAssignment(bundles[0]).get(); @@ -843,7 +844,7 @@ public void testRemoveNonExistBundleData() String topicToFindBundle = topicName + 0; NamespaceBundle bundleWillBeSplit = pulsar1.getNamespaceService().getBundle(TopicName.get(topicToFindBundle)); - String bundleDataPath = ModularLoadManagerImpl.BUNDLE_DATA_PATH + "/" + tenant + "/" + namespace; + String bundleDataPath = BundleDataResources.BUNDLE_DATA_BASE_PATH + "/" + tenant + "/" + namespace; CompletableFuture> children = bundlesCache.getChildren(bundleDataPath); List bundles = children.join(); assertTrue(bundles.contains(bundleWillBeSplit.getBundleRange())); From 693976a6e866f2bbf897dce81c4adcdc10620731 Mon Sep 17 00:00:00 2001 From: houxiaoyu Date: Mon, 4 Sep 2023 17:11:03 +0800 Subject: [PATCH 02/10] fix --- .../resources/BundleDataResourcesTest.java | 18 ++++++++++++++++++ .../broker/namespace/NamespaceServiceTest.java | 3 ++- 2 files changed, 20 insertions(+), 1 deletion(-) diff --git a/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/resources/BundleDataResourcesTest.java b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/resources/BundleDataResourcesTest.java index 64f9f7733e2c3..68ed651dab974 100644 --- a/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/resources/BundleDataResourcesTest.java +++ b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/resources/BundleDataResourcesTest.java @@ -1,3 +1,21 @@ +/* + * 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.broker.resources; import static org.apache.pulsar.broker.resources.BaseResources.joinPath; diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java index 03bb53eb9da24..ddf717b821061 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java @@ -60,6 +60,7 @@ import org.apache.pulsar.broker.loadbalance.impl.ModularLoadManagerWrapper; import org.apache.pulsar.broker.loadbalance.impl.SimpleResourceUnit; import org.apache.pulsar.broker.lookup.LookupResult; +import org.apache.pulsar.broker.resources.BundleDataResources; import org.apache.pulsar.broker.service.BrokerTestBase; import org.apache.pulsar.broker.service.Topic; import org.apache.pulsar.broker.service.persistent.PersistentTopic; @@ -649,7 +650,7 @@ public void testSplitBundleWithHighestThroughput() throws Exception { NamespaceBundle targetNamespaceBundle = bundles.findBundle(TopicName.get(topic + "0")); String bundle = targetNamespaceBundle.getBundleRange(); - String path = ModularLoadManagerImpl.getBundleDataPath(namespace + "/" + bundle); + String path = BundleDataResources.BUNDLE_DATA_BASE_PATH + "/" + bundle; NamespaceBundleStats defaultStats = new NamespaceBundleStats(); defaultStats.msgThroughputIn = 100000; defaultStats.msgThroughputOut = 100000; From 4c6a9bd3e40d82260df865e19186a2e91f0e7fcd Mon Sep 17 00:00:00 2001 From: houxiaoyu Date: Mon, 4 Sep 2023 18:12:28 +0800 Subject: [PATCH 03/10] fix --- .../pulsar/broker/resources/NamespaceResources.java | 1 - .../loadbalance/impl/ModularLoadManagerImpl.java | 12 ++++++++---- .../apache/pulsar/broker/admin/AdminApi2Test.java | 5 +++-- .../loadbalance/impl/ModularLoadManagerImplTest.java | 7 ++++--- .../broker/namespace/NamespaceServiceTest.java | 9 ++++----- .../pulsar/testclient/LoadSimulationController.java | 6 +++--- 6 files changed, 22 insertions(+), 18 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 982171cf3f92b..1ba353dccaa1c 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 @@ -52,7 +52,6 @@ public class NamespaceResources extends BaseResources { 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) { super(configurationStore, Policies.class, operationTimeoutSec); 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 9b7b1ae9758fb..dbc24eb880678 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 @@ -58,6 +58,7 @@ import org.apache.pulsar.broker.loadbalance.ModularLoadManager; import org.apache.pulsar.broker.loadbalance.ModularLoadManagerStrategy; import org.apache.pulsar.broker.loadbalance.impl.LoadManagerShared.BrokerTopicLoadingPredicate; +import org.apache.pulsar.broker.resources.PulsarResources; import org.apache.pulsar.broker.stats.prometheus.metrics.Summary; import org.apache.pulsar.client.admin.PulsarAdminException; import org.apache.pulsar.client.util.ExecutorProvider; @@ -168,6 +169,8 @@ public class ModularLoadManagerImpl implements ModularLoadManager { // Pulsar service used to initialize this. private PulsarService pulsar; + private PulsarResources pulsarResources; + // Executor service used to update broker data. private final ExecutorService executors; @@ -239,6 +242,7 @@ public boolean isEnableNonPersistentTopics(String brokerUrl) { @Override public void initialize(final PulsarService pulsar) { this.pulsar = pulsar; + this.pulsarResources = pulsar.getPulsarResources(); brokersData = pulsar.getCoordinationService().getLockManager(LocalBrokerData.class); resourceQuotaCache = pulsar.getLocalMetadataStore().getMetadataCache(ResourceQuota.class); timeAverageBrokerDataCache = pulsar.getLocalMetadataStore().getMetadataCache(TimeAverageBrokerData.class); @@ -268,7 +272,7 @@ public void initialize(final PulsarService pulsar) { LoadManagerShared.refreshBrokerToFailureDomainMap(pulsar, brokerToFailureDomainMap); // register listeners for domain changes - pulsar.getPulsarResources().getClusterResources().getFailureDomainResources() + pulsarResources.getClusterResources().getFailureDomainResources() .registerListener(__ -> { executors.execute( () -> LoadManagerShared.refreshBrokerToFailureDomainMap(pulsar, brokerToFailureDomainMap)); @@ -376,7 +380,7 @@ public CompletableFuture> getAvailableBrokersAsync() { public BundleData getBundleDataOrDefault(final String bundle) { BundleData bundleData = null; try { - Optional optBundleData = pulsar.getPulsarResources().getBundleDataResources().getBundleData(bundle).join(); + Optional optBundleData = pulsarResources.getBundleDataResources().getBundleData(bundle).join(); if (optBundleData.isPresent()) { return optBundleData.get(); } @@ -1141,7 +1145,7 @@ public void writeBundleDataOnZooKeeper() { for (Map.Entry entry : loadData.getBundleData().entrySet()) { final String bundle = entry.getKey(); final BundleData data = entry.getValue(); - futures.add(pulsar.getPulsarResources().getBundleDataResources().updateBundleData(bundle, data)); + futures.add(pulsarResources.getBundleDataResources().updateBundleData(bundle, data)); } // Write the time average broker data to metadata store. @@ -1162,7 +1166,7 @@ public void writeBundleDataOnZooKeeper() { private void deleteBundleDataFromMetadataStore(String bundle) { try { - pulsar.getPulsarResources().getBundleDataResources().deleteBundleData(bundle).join(); + pulsarResources.getBundleDataResources().deleteBundleData(bundle).join(); } catch (Exception e) { if (!(e.getCause() instanceof NotFoundException)) { log.warn("Failed to delete bundle-data {} from metadata store", bundle, e); 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 c68010f967b9b..071aa9b379918 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 @@ -21,6 +21,7 @@ import static java.util.concurrent.TimeUnit.MINUTES; import static org.apache.commons.lang3.StringUtils.isBlank; import static org.apache.pulsar.broker.BrokerTestUtil.newUniqueName; +import static org.apache.pulsar.broker.resources.BundleDataResources.BUNDLE_DATA_BASE_PATH; import static org.mockito.Mockito.spy; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; @@ -1671,7 +1672,7 @@ 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; + final String bundleDataPath = BUNDLE_DATA_BASE_PATH + "/" + tenant; assertFalse(pulsar.getLocalMetadataStore().exists(managedLedgersPath).join()); assertFalse(pulsar.getLocalMetadataStore().exists(partitionedTopicPath).join()); assertFalse(pulsar.getLocalMetadataStore().exists(localPoliciesPath).join()); @@ -1738,7 +1739,7 @@ public void testDeleteNamespace(NamespaceAttr namespaceAttr) throws Exception { assertFalse(admin.topics().getList(namespace).isEmpty()); final String managedLedgersPath = "/managed-ledgers/" + namespace; - final String bundleDataPath = "/loadbalance/bundle-data/" + namespace; + final String bundleDataPath = BUNDLE_DATA_BASE_PATH + "/" + namespace; // Trigger bundle owned by brokers. pulsarClient.newProducer().topic(topic).create().close(); // Trigger bundle data write to ZK. diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImplTest.java index 6bc8196ae6ecc..1699413fa52af 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImplTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImplTest.java @@ -20,6 +20,7 @@ import static java.lang.Thread.sleep; import static org.apache.pulsar.broker.loadbalance.impl.ModularLoadManagerImpl.TIME_AVERAGE_BROKER_ZPATH; +import static org.apache.pulsar.broker.resources.BundleDataResources.BUNDLE_DATA_BASE_PATH; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doReturn; @@ -291,7 +292,7 @@ public void testEvenBundleDistribution() throws Exception { final TimeAverageMessageData longTermMessageData = new TimeAverageMessageData(1000); longTermMessageData.setMsgRateIn(1000); bundleData.setLongTermData(longTermMessageData); - final String firstBundleDataPath = String.format("%s/%s", BundleDataResources.BUNDLE_DATA_BASE_PATH, bundles[0]); + final String firstBundleDataPath = String.format("%s/%s", BUNDLE_DATA_BASE_PATH, bundles[0]); // Write long message rate for first bundle to ensure that even bundle distribution is not a coincidence of // balancing by message rate. If we were balancing by message rate, one of the brokers should only have this // one bundle. @@ -387,7 +388,7 @@ public void testMaxTopicDistributionToBroker() throws Exception { final TimeAverageMessageData longTermMessageData = new TimeAverageMessageData(1000); longTermMessageData.setMsgRateIn(1000); bundleData.setLongTermData(longTermMessageData); - final String firstBundleDataPath = String.format("%s/%s", BundleDataResources.BUNDLE_DATA_BASE_PATH, bundles[0]); + final String firstBundleDataPath = String.format("%s/%s", BUNDLE_DATA_BASE_PATH, bundles[0]); pulsar1.getLocalMetadataStore().getMetadataCache(BundleData.class).create(firstBundleDataPath, bundleData).join(); String maxTopicOwnedBroker = primaryLoadManager.selectBrokerForAssignment(bundles[0]).get(); @@ -844,7 +845,7 @@ public void testRemoveNonExistBundleData() String topicToFindBundle = topicName + 0; NamespaceBundle bundleWillBeSplit = pulsar1.getNamespaceService().getBundle(TopicName.get(topicToFindBundle)); - String bundleDataPath = BundleDataResources.BUNDLE_DATA_BASE_PATH + "/" + tenant + "/" + namespace; + String bundleDataPath = BUNDLE_DATA_BASE_PATH + "/" + tenant + "/" + namespace; CompletableFuture> children = bundlesCache.getChildren(bundleDataPath); List bundles = children.join(); assertTrue(bundles.contains(bundleWillBeSplit.getBundleRange())); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java index ddf717b821061..9c17874daa0e9 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java @@ -18,6 +18,7 @@ */ package org.apache.pulsar.broker.namespace; +import static org.apache.pulsar.broker.resources.BundleDataResources.BUNDLE_DATA_BASE_PATH; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doReturn; @@ -650,7 +651,7 @@ public void testSplitBundleWithHighestThroughput() throws Exception { NamespaceBundle targetNamespaceBundle = bundles.findBundle(TopicName.get(topic + "0")); String bundle = targetNamespaceBundle.getBundleRange(); - String path = BundleDataResources.BUNDLE_DATA_BASE_PATH + "/" + bundle; + String path = BUNDLE_DATA_BASE_PATH + "/" + bundle; NamespaceBundleStats defaultStats = new NamespaceBundleStats(); defaultStats.msgThroughputIn = 100000; defaultStats.msgThroughputOut = 100000; @@ -692,7 +693,6 @@ public void testHeartbeatNamespaceMatch() throws Exception { @Test public void testModularLoadManagerRemoveInactiveBundleFromLoadData() throws Exception { - final String BUNDLE_DATA_PATH = "/loadbalance/bundle-data"; final String namespace = "pulsar/test/ns1"; final String topic1 = "persistent://" + namespace + "/topic1"; final String topic2 = "persistent://" + namespace + "/topic2"; @@ -743,13 +743,12 @@ public void testModularLoadManagerRemoveInactiveBundleFromLoadData() throws Exce Awaitility.await().untilAsserted(() -> { assertNull(loadData.getBundleData().get(oldBundle.toString())); - assertFalse(bundlesCache.exists(BUNDLE_DATA_PATH + "/" + oldBundle.toString()).get()); + assertFalse(bundlesCache.exists(BUNDLE_DATA_BASE_PATH + "/" + oldBundle.toString()).get()); }); } @Test public void testModularLoadManagerRemoveBundleAndLoad() throws Exception { - final String BUNDLE_DATA_PATH = "/loadbalance/bundle-data"; final String namespace = "prop/ns-abc"; final String bundleName = namespace + "/0x00000000_0xffffffff"; final String topic1 = "persistent://" + namespace + "/topic1"; @@ -784,7 +783,7 @@ public void testModularLoadManagerRemoveBundleAndLoad() throws Exception { pulsar.getBrokerService().updateRates(); waitResourceDataUpdateToZK(loadManager); - String path = BUNDLE_DATA_PATH + "/" + bundleName; + String path = BUNDLE_DATA_BASE_PATH + "/" + bundleName; Optional getResult = pulsar.getLocalMetadataStore().get(path).get(); assertTrue(getResult.isPresent()); diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/LoadSimulationController.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/LoadSimulationController.java index bbe535df5e289..da31b27f5add4 100644 --- a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/LoadSimulationController.java +++ b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/LoadSimulationController.java @@ -18,6 +18,7 @@ */ package org.apache.pulsar.testclient; +import static org.apache.pulsar.broker.resources.BundleDataResources.BUNDLE_DATA_BASE_PATH; import com.beust.jcommander.JCommander; import com.beust.jcommander.Parameter; import com.beust.jcommander.ParameterException; @@ -61,7 +62,6 @@ public class LoadSimulationController { private static final Logger log = LoggerFactory.getLogger(LoadSimulationController.class); private static final String QUOTA_ROOT = "/loadbalance/resource-quota/namespace"; - private static final String BUNDLE_DATA_ROOT = "/loadbalance/bundle-data"; // Input streams for each client to send commands through. private final DataInputStream[] inputStreams; @@ -427,7 +427,7 @@ private void handleCopy(final ShellArguments arguments) throws Exception { "/loadbalance/resource-quota/namespace/%s/%s/%s/0x00000000_0xffffffff", tenantName, cluster, mangledNamespace); final String newAPITargetPath = String.format( - "/loadbalance/bundle-data/%s/%s/%s/0x00000000_0xffffffff", tenantName, cluster, + "%s/%s/%s/%s/0x00000000_0xffffffff", BUNDLE_DATA_BASE_PATH, tenantName, cluster, mangledNamespace); try { ZkUtils.createFullPathOptimistic(targetZKClient, oldAPITargetPath, @@ -484,7 +484,7 @@ private void handleSimulate(final ShellArguments arguments) throws Exception { futures.add(threadPool.submit(() -> { for (final Map.Entry entry : bundleToQuota.entrySet()) { final String bundle = entry.getKey(); - final String newAPIPath = bundle.replace(QUOTA_ROOT, BUNDLE_DATA_ROOT); + final String newAPIPath = bundle.replace(QUOTA_ROOT, BUNDLE_DATA_BASE_PATH); final ResourceQuota quota = entry.getValue(); final int tenantStart = QUOTA_ROOT.length() + 1; final String topic = String.format("persistent://%s/t", bundle.substring(tenantStart)); From 3ce4fe2f29a3b930cf50e2f358a59f006abe4987 Mon Sep 17 00:00:00 2001 From: houxiaoyu Date: Mon, 4 Sep 2023 18:13:32 +0800 Subject: [PATCH 04/10] Fix test --- .../org/apache/pulsar/broker/resources/BundleDataResources.java | 1 - 1 file changed, 1 deletion(-) diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/BundleDataResources.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/BundleDataResources.java index e99f744896c4a..2bbee8b4ee425 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/BundleDataResources.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/BundleDataResources.java @@ -26,7 +26,6 @@ import org.apache.pulsar.policies.data.loadbalancer.BundleData; public class BundleDataResources extends BaseResources { - @VisibleForTesting public static final String BUNDLE_DATA_BASE_PATH = "/loadbalance/bundle-data"; public BundleDataResources(MetadataStore store, int operationTimeoutSec) { From ff5ee325dab2993ef07618d8f1fb76018af1da99 Mon Sep 17 00:00:00 2001 From: houxiaoyu Date: Mon, 4 Sep 2023 18:53:27 +0800 Subject: [PATCH 05/10] fix --- .../org/apache/pulsar/broker/resources/BundleDataResources.java | 1 - 1 file changed, 1 deletion(-) diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/BundleDataResources.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/BundleDataResources.java index 2bbee8b4ee425..1fa0ddba3b528 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/BundleDataResources.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/BundleDataResources.java @@ -18,7 +18,6 @@ */ package org.apache.pulsar.broker.resources; -import com.google.common.annotations.VisibleForTesting; import java.util.Optional; import java.util.concurrent.CompletableFuture; import org.apache.pulsar.common.naming.NamespaceName; From 5ae347c2b532487c2f60f79c2010d7a36b7ddbf2 Mon Sep 17 00:00:00 2001 From: houxiaoyu Date: Mon, 4 Sep 2023 19:14:05 +0800 Subject: [PATCH 06/10] fix checkstyle --- .../broker/loadbalance/impl/ModularLoadManagerImplTest.java | 1 - .../org/apache/pulsar/broker/namespace/NamespaceServiceTest.java | 1 - 2 files changed, 2 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImplTest.java index 1699413fa52af..522d242038e7d 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImplTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImplTest.java @@ -64,7 +64,6 @@ import org.apache.pulsar.broker.loadbalance.LoadData; import org.apache.pulsar.broker.loadbalance.LoadManager; import org.apache.pulsar.broker.loadbalance.impl.LoadManagerShared.BrokerTopicLoadingPredicate; -import org.apache.pulsar.broker.resources.BundleDataResources; import org.apache.pulsar.client.admin.Namespaces; import org.apache.pulsar.client.admin.PulsarAdmin; import org.apache.pulsar.client.admin.PulsarAdminException; diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java index 9c17874daa0e9..ca4bb64968702 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java @@ -61,7 +61,6 @@ import org.apache.pulsar.broker.loadbalance.impl.ModularLoadManagerWrapper; import org.apache.pulsar.broker.loadbalance.impl.SimpleResourceUnit; import org.apache.pulsar.broker.lookup.LookupResult; -import org.apache.pulsar.broker.resources.BundleDataResources; import org.apache.pulsar.broker.service.BrokerTestBase; import org.apache.pulsar.broker.service.Topic; import org.apache.pulsar.broker.service.persistent.PersistentTopic; From 46e18d5e064c37645f328fa62cbbe972f5064907 Mon Sep 17 00:00:00 2001 From: houxiaoyu Date: Mon, 4 Sep 2023 21:56:50 +0800 Subject: [PATCH 07/10] Fix --- .../org/apache/pulsar/broker/resources/PulsarResources.java | 2 +- 1 file changed, 1 insertion(+), 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 ba32ebc3b973f..c915e6834806c 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 @@ -62,7 +62,7 @@ public PulsarResources(MetadataStore localMetadataStore, MetadataStore configura if (configurationMetadataStore != null) { tenantResources = new TenantResources(configurationMetadataStore, operationTimeoutSec); clusterResources = new ClusterResources(configurationMetadataStore, operationTimeoutSec); - namespaceResources = new NamespaceResources(localMetadataStore, operationTimeoutSec); + namespaceResources = new NamespaceResources(configurationMetadataStore, operationTimeoutSec); resourcegroupResources = new ResourceGroupResources(configurationMetadataStore, operationTimeoutSec); } else { tenantResources = null; From e550412383d66e618d44ff365e0f2f9a03d51b1a Mon Sep 17 00:00:00 2001 From: houxiaoyu Date: Mon, 4 Sep 2023 23:59:03 +0800 Subject: [PATCH 08/10] fix test --- .../apache/pulsar/broker/testcontext/PulsarTestContext.java | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) 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 db09465dc10ad..c927a2e61d85e 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 @@ -712,8 +712,7 @@ protected void initializePulsarServices(SpyConfig spyConfig, Builder builder) { if (metadataStore == null) { metadataStore = builder.configurationMetadataStore; } - NamespaceResources nsr = spyConfigPulsarResources.spy(NamespaceResources.class, - builder.localMetadataStore, metadataStore, 30); + NamespaceResources nsr = spyConfigPulsarResources.spy(NamespaceResources.class,metadataStore, 30); TopicResources tsr = spyConfigPulsarResources.spy(TopicResources.class, metadataStore); pulsarResources( spyConfigPulsarResources.spy( From 22efeefbccdd9ef46f470db6b8902e6d0f631a0b Mon Sep 17 00:00:00 2001 From: houxiaoyu Date: Tue, 5 Sep 2023 10:31:05 +0800 Subject: [PATCH 09/10] Fix test --- .../apache/pulsar/broker/namespace/NamespaceServiceTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java index ca4bb64968702..ba665ea719564 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java @@ -650,7 +650,7 @@ public void testSplitBundleWithHighestThroughput() throws Exception { NamespaceBundle targetNamespaceBundle = bundles.findBundle(TopicName.get(topic + "0")); String bundle = targetNamespaceBundle.getBundleRange(); - String path = BUNDLE_DATA_BASE_PATH + "/" + bundle; + String path = BUNDLE_DATA_BASE_PATH + "/" + namespace + "/" + bundle; NamespaceBundleStats defaultStats = new NamespaceBundleStats(); defaultStats.msgThroughputIn = 100000; defaultStats.msgThroughputOut = 100000; From 0fda81b7f1b0871984a67e07d2ea3254c09dbd38 Mon Sep 17 00:00:00 2001 From: houxiaoyu Date: Tue, 5 Sep 2023 13:07:43 +0800 Subject: [PATCH 10/10] LoadBalanceResources --- .../broker/resources/BundleDataResources.java | 62 ---------------- .../resources/LoadBalanceResources.java | 72 +++++++++++++++++++ .../broker/resources/PulsarResources.java | 6 +- ...est.java => LoadBalanceResourcesTest.java} | 12 ++-- .../broker/admin/impl/NamespacesBase.java | 3 +- .../pulsar/broker/admin/impl/TenantsBase.java | 2 +- .../impl/ModularLoadManagerImpl.java | 8 ++- .../pulsar/broker/web/PulsarWebResource.java | 6 +- .../pulsar/broker/admin/AdminApi2Test.java | 2 +- .../impl/ModularLoadManagerImplTest.java | 2 +- .../namespace/NamespaceServiceTest.java | 2 +- .../testclient/LoadSimulationController.java | 2 +- 12 files changed, 96 insertions(+), 83 deletions(-) delete mode 100644 pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/BundleDataResources.java create mode 100644 pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/LoadBalanceResources.java rename pulsar-broker-common/src/test/java/org/apache/pulsar/broker/resources/{BundleDataResourcesTest.java => LoadBalanceResourcesTest.java} (83%) diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/BundleDataResources.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/BundleDataResources.java deleted file mode 100644 index 1fa0ddba3b528..0000000000000 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/BundleDataResources.java +++ /dev/null @@ -1,62 +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.broker.resources; - -import java.util.Optional; -import java.util.concurrent.CompletableFuture; -import org.apache.pulsar.common.naming.NamespaceName; -import org.apache.pulsar.metadata.api.MetadataStore; -import org.apache.pulsar.policies.data.loadbalancer.BundleData; - -public class BundleDataResources extends BaseResources { - public static final String BUNDLE_DATA_BASE_PATH = "/loadbalance/bundle-data"; - - public BundleDataResources(MetadataStore store, int operationTimeoutSec) { - super(store, BundleData.class, operationTimeoutSec); - } - - public CompletableFuture> getBundleData(String bundle) { - return getAsync(getBundleDataPath(bundle)); - } - - public CompletableFuture updateBundleData(String bundle, BundleData data) { - return setWithCreateAsync(getBundleDataPath(bundle), __ -> data); - } - - public CompletableFuture deleteBundleData(String bundle) { - return deleteAsync(getBundleDataPath(bundle)); - } - - // 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); - } - - // 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); - } - - // Get the metadata store path for the given bundle full name. - private String getBundleDataPath(final String bundle) { - return BUNDLE_DATA_BASE_PATH + "/" + bundle; - } -} diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/LoadBalanceResources.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/LoadBalanceResources.java new file mode 100644 index 0000000000000..839997a7035fe --- /dev/null +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/LoadBalanceResources.java @@ -0,0 +1,72 @@ +/* + * 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.broker.resources; + +import java.util.Optional; +import java.util.concurrent.CompletableFuture; +import lombok.Getter; +import org.apache.pulsar.common.naming.NamespaceName; +import org.apache.pulsar.metadata.api.MetadataStore; +import org.apache.pulsar.policies.data.loadbalancer.BundleData; + +@Getter +public class LoadBalanceResources { + public static final String BUNDLE_DATA_BASE_PATH = "/loadbalance/bundle-data"; + + private final BundleDataResources bundleDataResources; + + public LoadBalanceResources(MetadataStore store, int operationTimeoutSec) { + bundleDataResources = new BundleDataResources(store, operationTimeoutSec); + } + + public static class BundleDataResources extends BaseResources { + public BundleDataResources(MetadataStore store, int operationTimeoutSec) { + super(store, BundleData.class, operationTimeoutSec); + } + + public CompletableFuture> getBundleData(String bundle) { + return getAsync(getBundleDataPath(bundle)); + } + + public CompletableFuture updateBundleData(String bundle, BundleData data) { + return setWithCreateAsync(getBundleDataPath(bundle), __ -> data); + } + + public CompletableFuture deleteBundleData(String bundle) { + return deleteAsync(getBundleDataPath(bundle)); + } + + // 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); + } + + // 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); + } + + // Get the metadata store path for the given bundle full name. + private String getBundleDataPath(final String bundle) { + return BUNDLE_DATA_BASE_PATH + "/" + bundle; + } + } +} 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 c915e6834806c..ad872a5356cf4 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 @@ -48,7 +48,7 @@ public class PulsarResources { @Getter private final TopicResources topicResources; @Getter - private final BundleDataResources bundleDataResources; + private final LoadBalanceResources loadBalanceResources; @Getter private final Optional localMetadataStore; @Getter @@ -77,14 +77,14 @@ public PulsarResources(MetadataStore localMetadataStore, MetadataStore configura loadReportResources = new LoadManagerReportResources(localMetadataStore, operationTimeoutSec); bookieResources = new BookieResources(localMetadataStore, operationTimeoutSec); topicResources = new TopicResources(localMetadataStore); - bundleDataResources = new BundleDataResources(localMetadataStore, operationTimeoutSec); + loadBalanceResources = new LoadBalanceResources(localMetadataStore, operationTimeoutSec); } else { dynamicConfigResources = null; localPolicies = null; loadReportResources = null; bookieResources = null; topicResources = null; - bundleDataResources = null; + loadBalanceResources = null; } this.localMetadataStore = Optional.ofNullable(localMetadataStore); diff --git a/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/resources/BundleDataResourcesTest.java b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/resources/LoadBalanceResourcesTest.java similarity index 83% rename from pulsar-broker-common/src/test/java/org/apache/pulsar/broker/resources/BundleDataResourcesTest.java rename to pulsar-broker-common/src/test/java/org/apache/pulsar/broker/resources/LoadBalanceResourcesTest.java index 68ed651dab974..cd7dd01b66576 100644 --- a/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/resources/BundleDataResourcesTest.java +++ b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/resources/LoadBalanceResourcesTest.java @@ -19,7 +19,7 @@ package org.apache.pulsar.broker.resources; import static org.apache.pulsar.broker.resources.BaseResources.joinPath; -import static org.apache.pulsar.broker.resources.BundleDataResources.BUNDLE_DATA_BASE_PATH; +import static org.apache.pulsar.broker.resources.LoadBalanceResources.BUNDLE_DATA_BASE_PATH; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; import static org.testng.Assert.assertThrows; @@ -28,16 +28,16 @@ import org.testng.annotations.BeforeMethod; import org.testng.annotations.Test; -public class BundleDataResourcesTest { +public class LoadBalanceResourcesTest { private MetadataStore configurationStore; private MetadataStore localStore; - private BundleDataResources bundleDataResources; + private LoadBalanceResources loadBalanceResources; @BeforeMethod public void setup() { localStore = mock(MetadataStore.class); configurationStore = mock(MetadataStore.class); - bundleDataResources = new BundleDataResources(localStore, 30); + loadBalanceResources = new LoadBalanceResources(localStore, 30); } /** @@ -47,11 +47,11 @@ public void setup() { public void testDeleteBundleDataAsync() { NamespaceName ns = NamespaceName.get("my-tenant/my-ns"); String namespaceBundlePath = joinPath(BUNDLE_DATA_BASE_PATH, ns.toString()); - bundleDataResources.deleteBundleDataAsync(ns); + loadBalanceResources.getBundleDataResources().deleteBundleDataAsync(ns); String tenant="my-tenant"; String tenantBundlePath = joinPath(BUNDLE_DATA_BASE_PATH, tenant); - bundleDataResources.deleteBundleDataTenantAsync(tenant); + loadBalanceResources.getBundleDataResources().deleteBundleDataTenantAsync(tenant); verify(localStore).deleteRecursive(namespaceBundlePath); verify(localStore).deleteRecursive(tenantBundlePath); 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 index f7f0d18735f0c..8ab1f4dc86002 100644 --- 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 @@ -468,7 +468,8 @@ protected CompletableFuture internalClearZkSources() { // clear z-node of local policies .thenCompose(ignore -> getLocalPolicies().deleteLocalPoliciesAsync(namespaceName)) // clear /loadbalance/bundle-data - .thenCompose(ignore -> bundleDataResources().deleteBundleDataAsync(namespaceName)); + .thenCompose(ignore -> + loadBalanceResources().getBundleDataResources().deleteBundleDataAsync(namespaceName)); } 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 index 6f18c55b4e89c..74c0367e0b97c 100644 --- 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 @@ -236,7 +236,7 @@ protected CompletableFuture internalDeleteTenantAsync(String tenant) { .getPartitionedTopicResources().clearPartitionedTopicTenantAsync(tenant)) .thenCompose(__ -> pulsar().getPulsarResources().getLocalPolicies() .deleteLocalPoliciesTenantAsync(tenant)) - .thenCompose(__ -> pulsar().getPulsarResources().getBundleDataResources() + .thenCompose(__ -> pulsar().getPulsarResources().getLoadBalanceResources().getBundleDataResources() .deleteBundleDataTenantAsync(tenant)); } 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 dbc24eb880678..586478efa50f7 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 @@ -380,7 +380,8 @@ public CompletableFuture> getAvailableBrokersAsync() { public BundleData getBundleDataOrDefault(final String bundle) { BundleData bundleData = null; try { - Optional optBundleData = pulsarResources.getBundleDataResources().getBundleData(bundle).join(); + Optional optBundleData = + pulsarResources.getLoadBalanceResources().getBundleDataResources().getBundleData(bundle).join(); if (optBundleData.isPresent()) { return optBundleData.get(); } @@ -1145,7 +1146,8 @@ public void writeBundleDataOnZooKeeper() { for (Map.Entry entry : loadData.getBundleData().entrySet()) { final String bundle = entry.getKey(); final BundleData data = entry.getValue(); - futures.add(pulsarResources.getBundleDataResources().updateBundleData(bundle, data)); + futures.add( + pulsarResources.getLoadBalanceResources().getBundleDataResources().updateBundleData(bundle, data)); } // Write the time average broker data to metadata store. @@ -1166,7 +1168,7 @@ public void writeBundleDataOnZooKeeper() { private void deleteBundleDataFromMetadataStore(String bundle) { try { - pulsarResources.getBundleDataResources().deleteBundleData(bundle).join(); + pulsarResources.getLoadBalanceResources().getBundleDataResources().deleteBundleData(bundle).join(); } catch (Exception e) { if (!(e.getCause() instanceof NotFoundException)) { log.warn("Failed to delete bundle-data {} from metadata store", bundle, e); 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 2d7247acc27f4..927a5b92780bc 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 @@ -60,9 +60,9 @@ import org.apache.pulsar.broker.namespace.LookupOptions; import org.apache.pulsar.broker.namespace.NamespaceService; import org.apache.pulsar.broker.resources.BookieResources; -import org.apache.pulsar.broker.resources.BundleDataResources; import org.apache.pulsar.broker.resources.ClusterResources; import org.apache.pulsar.broker.resources.DynamicConfigurationResources; +import org.apache.pulsar.broker.resources.LoadBalanceResources; import org.apache.pulsar.broker.resources.LocalPoliciesResources; import org.apache.pulsar.broker.resources.NamespaceResources; import org.apache.pulsar.broker.resources.NamespaceResources.IsolationPolicyResources; @@ -1112,8 +1112,8 @@ protected NamespaceResources namespaceResources() { return pulsar().getPulsarResources().getNamespaceResources(); } - protected BundleDataResources bundleDataResources() { - return pulsar().getPulsarResources().getBundleDataResources(); + protected LoadBalanceResources loadBalanceResources() { + return pulsar().getPulsarResources().getLoadBalanceResources(); } protected ResourceGroupResources resourceGroupResources() { 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 071aa9b379918..5abb0e02e588b 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 @@ -21,7 +21,7 @@ import static java.util.concurrent.TimeUnit.MINUTES; import static org.apache.commons.lang3.StringUtils.isBlank; import static org.apache.pulsar.broker.BrokerTestUtil.newUniqueName; -import static org.apache.pulsar.broker.resources.BundleDataResources.BUNDLE_DATA_BASE_PATH; +import static org.apache.pulsar.broker.resources.LoadBalanceResources.BUNDLE_DATA_BASE_PATH; import static org.mockito.Mockito.spy; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImplTest.java index 522d242038e7d..557393682fb03 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImplTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImplTest.java @@ -20,7 +20,7 @@ import static java.lang.Thread.sleep; import static org.apache.pulsar.broker.loadbalance.impl.ModularLoadManagerImpl.TIME_AVERAGE_BROKER_ZPATH; -import static org.apache.pulsar.broker.resources.BundleDataResources.BUNDLE_DATA_BASE_PATH; +import static org.apache.pulsar.broker.resources.LoadBalanceResources.BUNDLE_DATA_BASE_PATH; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doReturn; diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java index ba665ea719564..2e584489c0675 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java @@ -18,7 +18,7 @@ */ package org.apache.pulsar.broker.namespace; -import static org.apache.pulsar.broker.resources.BundleDataResources.BUNDLE_DATA_BASE_PATH; +import static org.apache.pulsar.broker.resources.LoadBalanceResources.BUNDLE_DATA_BASE_PATH; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doReturn; diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/LoadSimulationController.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/LoadSimulationController.java index da31b27f5add4..f2ccd82b3901a 100644 --- a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/LoadSimulationController.java +++ b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/LoadSimulationController.java @@ -18,7 +18,7 @@ */ package org.apache.pulsar.testclient; -import static org.apache.pulsar.broker.resources.BundleDataResources.BUNDLE_DATA_BASE_PATH; +import static org.apache.pulsar.broker.resources.LoadBalanceResources.BUNDLE_DATA_BASE_PATH; import com.beust.jcommander.JCommander; import com.beust.jcommander.Parameter; import com.beust.jcommander.ParameterException;