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/NamespaceResources.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/NamespaceResources.java index b5ccc9a5a9077..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 @@ -49,18 +49,15 @@ 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 +376,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..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,6 +48,8 @@ public class PulsarResources { @Getter private final TopicResources topicResources; @Getter + private final LoadBalanceResources loadBalanceResources; + @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(configurationMetadataStore, 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); + loadBalanceResources = new LoadBalanceResources(localMetadataStore, operationTimeoutSec); } else { dynamicConfigResources = null; localPolicies = null; loadReportResources = null; bookieResources = null; topicResources = null; + loadBalanceResources = null; } this.localMetadataStore = Optional.ofNullable(localMetadataStore); diff --git a/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/resources/LoadBalanceResourcesTest.java b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/resources/LoadBalanceResourcesTest.java new file mode 100644 index 0000000000000..cd7dd01b66576 --- /dev/null +++ b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/resources/LoadBalanceResourcesTest.java @@ -0,0 +1,62 @@ +/* + * 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; +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; +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 LoadBalanceResourcesTest { + private MetadataStore configurationStore; + private MetadataStore localStore; + private LoadBalanceResources loadBalanceResources; + + @BeforeMethod + public void setup() { + localStore = mock(MetadataStore.class); + configurationStore = mock(MetadataStore.class); + loadBalanceResources = new LoadBalanceResources(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()); + loadBalanceResources.getBundleDataResources().deleteBundleDataAsync(ns); + + String tenant="my-tenant"; + String tenantBundlePath = joinPath(BUNDLE_DATA_BASE_PATH, tenant); + loadBalanceResources.getBundleDataResources().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..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 -> namespaceResources().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 b93f3e3c6ebcc..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().getNamespaceResources() + .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 0d5dbf489e90f..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 @@ -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; @@ -91,9 +92,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 +118,6 @@ public class ModularLoadManagerImpl implements ModularLoadManager { private LockManager brokersData; private ResourceLock brokerDataLock; - private MetadataCache bundlesCache; private MetadataCache resourceQuotaCache; private MetadataCache timeAverageBrokerDataCache; @@ -172,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; @@ -243,8 +242,8 @@ 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); - bundlesCache = pulsar.getLocalMetadataStore().getMetadataCache(BundleData.class); resourceQuotaCache = pulsar.getLocalMetadataStore().getMetadataCache(ResourceQuota.class); timeAverageBrokerDataCache = pulsar.getLocalMetadataStore().getMetadataCache(TimeAverageBrokerData.class); pulsar.getLocalMetadataStore().registerListener(this::handleDataNotification); @@ -273,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)); @@ -381,7 +380,8 @@ public CompletableFuture> getAvailableBrokersAsync() { public BundleData getBundleDataOrDefault(final String bundle) { BundleData bundleData = null; try { - Optional optBundleData = bundlesCache.get(getBundleDataPath(bundle)).join(); + Optional optBundleData = + pulsarResources.getLoadBalanceResources().getBundleDataResources().getBundleData(bundle).join(); if (optBundleData.isPresent()) { return optBundleData.get(); } @@ -418,11 +418,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 +1146,8 @@ 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( + pulsarResources.getLoadBalanceResources().getBundleDataResources().updateBundleData(bundle, data)); } // Write the time average broker data to metadata store. @@ -1173,7 +1168,7 @@ public void writeBundleDataOnZooKeeper() { private void deleteBundleDataFromMetadataStore(String bundle) { try { - bundlesCache.delete(getBundleDataPath(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 e65ef50c72aff..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 @@ -62,6 +62,7 @@ import org.apache.pulsar.broker.resources.BookieResources; 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; @@ -1111,6 +1112,10 @@ protected NamespaceResources namespaceResources() { return pulsar().getPulsarResources().getNamespaceResources(); } + protected LoadBalanceResources loadBalanceResources() { + return pulsar().getPulsarResources().getLoadBalanceResources(); + } + protected ResourceGroupResources resourceGroupResources() { return pulsar().getPulsarResources().getResourcegroupResources(); } 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..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,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.LoadBalanceResources.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 d8acb6d24e9ef..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,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.LoadBalanceResources.BUNDLE_DATA_BASE_PATH; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doReturn; @@ -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", 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", 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 = 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 03bb53eb9da24..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,6 +18,7 @@ */ package org.apache.pulsar.broker.namespace; +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; @@ -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 = BUNDLE_DATA_BASE_PATH + "/" + namespace + "/" + bundle; NamespaceBundleStats defaultStats = new NamespaceBundleStats(); defaultStats.msgThroughputIn = 100000; defaultStats.msgThroughputOut = 100000; @@ -691,7 +692,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"; @@ -742,13 +742,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"; @@ -783,7 +782,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-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( 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..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,6 +18,7 @@ */ package org.apache.pulsar.testclient; +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; @@ -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));