You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2017/10/30 04:57:15 UTC
[GitHub] rdhabalia closed pull request #869: Broker-stats return load-report based on loadmanager + fix: modularLoadManager-loadReport
rdhabalia closed pull request #869: Broker-stats return load-report based on loadmanager + fix: modularLoadManager-loadReport
URL: https://github.com/apache/incubator-pulsar/pull/869
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/BrokerData.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/BrokerData.java
index 13b381a3c..eaccec850 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/BrokerData.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/BrokerData.java
@@ -21,6 +21,8 @@
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+import org.apache.pulsar.policies.data.loadbalancer.LocalBrokerData;
+
/**
* Data class containing three components comprising all the data available for the leader broker about other brokers: -
* The local broker data which is written to ZooKeeper by each individual broker (LocalBrokerData). - The time average
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/BundleData.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/BundleData.java
index c523060f7..914c57f98 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/BundleData.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/BundleData.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.broker;
+import org.apache.pulsar.policies.data.loadbalancer.JSONWritable;
import org.apache.pulsar.policies.data.loadbalancer.NamespaceBundleStats;
/**
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/TimeAverageBrokerData.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/TimeAverageBrokerData.java
index 3a3d5561d..05f93a890 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/TimeAverageBrokerData.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/TimeAverageBrokerData.java
@@ -21,6 +21,7 @@
import java.util.Map;
import java.util.Set;
+import org.apache.pulsar.policies.data.loadbalancer.JSONWritable;
import org.apache.pulsar.policies.data.loadbalancer.NamespaceBundleStats;
/**
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/BrokerStats.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/BrokerStats.java
index d3d5581a8..913ee78d3 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/BrokerStats.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/BrokerStats.java
@@ -42,6 +42,7 @@
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.stats.AllocatorStats;
import org.apache.pulsar.common.stats.Metrics;
+import org.apache.pulsar.policies.data.loadbalancer.LoadManagerReport;
import org.apache.pulsar.policies.data.loadbalancer.LoadReport;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -151,7 +152,7 @@ public AllocatorStats getAllocatorStats(@PathParam("allocator") String allocator
@Path("/load-report")
@ApiOperation(value = "Get Load for this broker", notes = "consists of destinationstats & systemResourceUsage", response = LoadReport.class)
@ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission") })
- public LoadReport getLoadReport() throws Exception {
+ public LoadManagerReport getLoadReport() throws Exception {
// Ensure super user access only
validateSuperUserAccess();
try {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LoadManager.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LoadManager.java
index a10432884..f9afae439 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LoadManager.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LoadManager.java
@@ -18,7 +18,6 @@
*/
package org.apache.pulsar.broker.loadbalance;
-import java.io.IOException;
import java.util.List;
import org.apache.pulsar.broker.PulsarServerException;
@@ -28,9 +27,8 @@
import org.apache.pulsar.broker.loadbalance.impl.SimpleLoadManagerImpl;
import org.apache.pulsar.common.naming.ServiceUnitId;
import org.apache.pulsar.common.stats.Metrics;
-import org.apache.pulsar.policies.data.loadbalancer.LoadReport;
+import org.apache.pulsar.policies.data.loadbalancer.LoadManagerReport;
import org.apache.pulsar.policies.data.loadbalancer.ServiceLookupData;
-import org.apache.pulsar.policies.data.loadbalancer.SystemResourceUsage;
import org.apache.pulsar.zookeeper.ZooKeeperCache.Deserializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -62,7 +60,7 @@
/**
* Generate the load report
*/
- LoadReport generateLoadReport() throws Exception;
+ LoadManagerReport generateLoadReport() throws Exception;
/**
* Returns {@link Deserializer} to deserialize load report
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/ModularLoadManager.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/ModularLoadManager.java
index 308da9e93..e07bc5a78 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/ModularLoadManager.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/ModularLoadManager.java
@@ -21,6 +21,7 @@
import org.apache.pulsar.broker.PulsarServerException;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.common.naming.ServiceUnitId;
+import org.apache.pulsar.policies.data.loadbalancer.LocalBrokerData;
import org.apache.pulsar.policies.data.loadbalancer.ServiceLookupData;
import org.apache.pulsar.zookeeper.ZooKeeperCache.Deserializer;
@@ -82,7 +83,7 @@
/**
* As any broker, retrieve the namespace bundle stats and system resource usage to update data local to this broker.
*/
- void updateLocalBrokerData();
+ LocalBrokerData updateLocalBrokerData();
/**
* As any broker, write the local broker data to ZooKeeper.
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/BundleSplitterTask.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/BundleSplitterTask.java
index 71578a3d6..117430abc 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/BundleSplitterTask.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/BundleSplitterTask.java
@@ -24,8 +24,6 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-
-import org.apache.pulsar.broker.LocalBrokerData;
//import org.apache.pulsar.broker.MessageData;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
@@ -33,6 +31,7 @@
import org.apache.pulsar.broker.loadbalance.BundleSplitStrategy;
import org.apache.pulsar.broker.loadbalance.LoadData;
import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.policies.data.loadbalancer.LocalBrokerData;
import org.apache.pulsar.policies.data.loadbalancer.NamespaceBundleStats;
/**
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/LeastLongTermMessageRate.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/LeastLongTermMessageRate.java
index c04f9ec6a..ac2c9d221 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/LeastLongTermMessageRate.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/LeastLongTermMessageRate.java
@@ -24,12 +24,12 @@
import org.apache.pulsar.broker.BrokerData;
import org.apache.pulsar.broker.BundleData;
-import org.apache.pulsar.broker.LocalBrokerData;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.TimeAverageBrokerData;
import org.apache.pulsar.broker.TimeAverageMessageData;
import org.apache.pulsar.broker.loadbalance.LoadData;
import org.apache.pulsar.broker.loadbalance.ModularLoadManagerStrategy;
+import org.apache.pulsar.policies.data.loadbalancer.LocalBrokerData;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
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 c33e1763e..021bb18cf 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
@@ -38,7 +38,6 @@
import org.apache.commons.lang3.SystemUtils;
import org.apache.pulsar.broker.BrokerData;
import org.apache.pulsar.broker.BundleData;
-import org.apache.pulsar.broker.LocalBrokerData;
import org.apache.pulsar.broker.PulsarServerException;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
@@ -47,6 +46,7 @@
import org.apache.pulsar.broker.loadbalance.BrokerFilter;
import org.apache.pulsar.broker.loadbalance.BrokerFilterException;
import org.apache.pulsar.broker.loadbalance.BrokerHostUsage;
+import org.apache.pulsar.broker.loadbalance.BundleSplitStrategy;
import org.apache.pulsar.broker.loadbalance.LoadData;
import org.apache.pulsar.broker.loadbalance.LoadManager;
import org.apache.pulsar.broker.loadbalance.LoadSheddingStrategy;
@@ -58,6 +58,7 @@
import org.apache.pulsar.common.naming.ServiceUnitId;
import org.apache.pulsar.common.policies.data.ResourceQuota;
import org.apache.pulsar.common.util.ObjectMapperFactory;
+import org.apache.pulsar.policies.data.loadbalancer.LocalBrokerData;
import org.apache.pulsar.policies.data.loadbalancer.NamespaceBundleStats;
import org.apache.pulsar.policies.data.loadbalancer.SystemResourceUsage;
import org.apache.pulsar.zookeeper.ZooKeeperCache.Deserializer;
@@ -66,16 +67,13 @@
import org.apache.pulsar.zookeeper.ZooKeeperDataCache;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.KeeperException.NoNodeException;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.apache.pulsar.broker.loadbalance.BundleSplitStrategy;
-
-import org.apache.zookeeper.KeeperException.NoNodeException;
-
import io.netty.util.concurrent.DefaultThreadFactory;
public class ModularLoadManagerImpl implements ModularLoadManager, ZooKeeperCacheListener<LocalBrokerData> {
@@ -760,15 +758,17 @@ public void stop() throws PulsarServerException {
/**
* As any broker, retrieve the namespace bundle stats and system resource usage to update data local to this broker.
+ * @return
*/
@Override
- public void updateLocalBrokerData() {
+ public LocalBrokerData updateLocalBrokerData() {
try {
final SystemResourceUsage systemResourceUsage = LoadManagerShared.getSystemResourceUsage(brokerHostUsage);
localData.update(systemResourceUsage, getBundleStats());
} catch (Exception e) {
log.warn("Error when attempting to update local broker data: {}", e);
}
+ return localData;
}
/**
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerWrapper.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerWrapper.java
index a1f831848..ed9199159 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerWrapper.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerWrapper.java
@@ -28,7 +28,7 @@
import org.apache.pulsar.broker.loadbalance.ResourceUnit;
import org.apache.pulsar.common.naming.ServiceUnitId;
import org.apache.pulsar.common.stats.Metrics;
-import org.apache.pulsar.policies.data.loadbalancer.LoadReport;
+import org.apache.pulsar.policies.data.loadbalancer.LoadManagerReport;
import org.apache.pulsar.policies.data.loadbalancer.ServiceLookupData;
import org.apache.pulsar.zookeeper.ZooKeeperCache.Deserializer;
@@ -58,9 +58,8 @@ public void doNamespaceBundleSplit() {
}
@Override
- public LoadReport generateLoadReport() {
- loadManager.updateLocalBrokerData();
- return null;
+ public LoadManagerReport generateLoadReport() {
+ return loadManager.updateLocalBrokerData();
}
@Override
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/OverloadShedder.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/OverloadShedder.java
index 203f0a3a5..341136dc4 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/OverloadShedder.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/OverloadShedder.java
@@ -23,11 +23,11 @@
import org.apache.pulsar.broker.BrokerData;
import org.apache.pulsar.broker.BundleData;
-import org.apache.pulsar.broker.LocalBrokerData;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.TimeAverageMessageData;
import org.apache.pulsar.broker.loadbalance.LoadData;
import org.apache.pulsar.broker.loadbalance.LoadSheddingStrategy;
+import org.apache.pulsar.policies.data.loadbalancer.LocalBrokerData;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java
index 86b83a9f7..d379daaaf 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java
@@ -128,7 +128,7 @@ public void setup() throws Exception {
// create otherbroker to test redirect on calls that need
// namespace ownership
- mockPulsarSetup = new MockedPulsarService(this.conf, this.pulsar, this.admin);
+ mockPulsarSetup = new MockedPulsarService(this.conf);
mockPulsarSetup.setup();
otherPulsar = mockPulsarSetup.getPulsar();
otheradmin = mockPulsarSetup.getAdmin();
@@ -1701,20 +1701,15 @@ public void failed(Throwable e) {
static class MockedPulsarService extends MockedPulsarServiceBaseTest {
private ServiceConfiguration conf;
- private PulsarService pulsar;
- private PulsarAdmin admin;
- public MockedPulsarService(ServiceConfiguration conf, PulsarService pulsar, PulsarAdmin admin) {
+ public MockedPulsarService(ServiceConfiguration conf) {
super();
this.conf = conf;
- this.pulsar = pulsar;
- this.admin = admin;
}
@Override
protected void setup() throws Exception {
- conf.setLoadBalancerEnabled(false);
- conf.setClusterName("test");
+ super.conf.setLoadManagerClassName(conf.getLoadManagerClassName());
super.internalSetup();
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest2.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest2.java
index 8278f11b3..c76703f75 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest2.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest2.java
@@ -34,6 +34,8 @@
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.admin.AdminApiTest.MockedPulsarService;
import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
+import org.apache.pulsar.broker.loadbalance.impl.ModularLoadManagerImpl;
+import org.apache.pulsar.broker.loadbalance.impl.SimpleLoadManagerImpl;
import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.client.admin.PulsarAdmin;
@@ -59,8 +61,10 @@
import org.apache.pulsar.common.policies.data.PersistentTopicStats;
import org.apache.pulsar.common.policies.data.PropertyAdmin;
import org.apache.pulsar.common.policies.data.RetentionPolicies;
+import org.apache.pulsar.policies.data.loadbalancer.LoadManagerReport;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.testng.Assert;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.DataProvider;
@@ -70,6 +74,7 @@
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.google.common.hash.Hashing;
+import com.google.gson.JsonObject;
public class AdminApiTest2 extends MockedPulsarServiceBaseTest {
@@ -77,11 +82,6 @@
private MockedPulsarService mockPulsarSetup;
- private PulsarService otherPulsar;
-
- private PulsarAdmin otheradmin;
-
- private NamespaceBundleFactory bundleFactory;
@BeforeMethod
@Override
@@ -89,14 +89,10 @@ public void setup() throws Exception {
conf.setLoadBalancerEnabled(true);
super.internalSetup();
- bundleFactory = new NamespaceBundleFactory(pulsar, Hashing.crc32());
-
// create otherbroker to test redirect on calls that need
// namespace ownership
- mockPulsarSetup = new MockedPulsarService(this.conf, this.pulsar, this.admin);
+ mockPulsarSetup = new MockedPulsarService(this.conf);
mockPulsarSetup.setup();
- otherPulsar = mockPulsarSetup.getPulsar();
- otheradmin = mockPulsarSetup.getAdmin();
// Setup namespaces
admin.clusters().createCluster("use", new ClusterData("http://127.0.0.1" + ":" + BROKER_WEBSERVICE_PORT));
@@ -524,4 +520,34 @@ private void publishMessagesOnPersistentTopic(String topicName, int messages, in
producer.close();
}
+ /**
+ * It verifies that pulsar with different load-manager generates different load-report and returned by admin-api
+ *
+ * @throws Exception
+ */
+ @Test
+ public void testLoadReportApi() throws Exception {
+
+ this.conf.setLoadManagerClassName(SimpleLoadManagerImpl.class.getName());
+ MockedPulsarService mockPulsarSetup1 = new MockedPulsarService(this.conf);
+ mockPulsarSetup1.setup();
+ PulsarService simpleLoadManager = mockPulsarSetup1.getPulsar();
+ PulsarAdmin simpleLoadManagerAdmin = mockPulsarSetup1.getAdmin();
+ assertNotNull(simpleLoadManagerAdmin.brokerStats().getLoadReport());
+
+ this.conf.setLoadManagerClassName(ModularLoadManagerImpl.class.getName());
+ MockedPulsarService mockPulsarSetup2 = new MockedPulsarService(this.conf);
+ mockPulsarSetup2.setup();
+ PulsarService modularLoadManager = mockPulsarSetup2.getPulsar();
+ PulsarAdmin modularLoadManagerAdmin = mockPulsarSetup2.getAdmin();
+ assertNotNull(modularLoadManagerAdmin.brokerStats().getLoadReport());
+
+ simpleLoadManagerAdmin.close();
+ simpleLoadManager.close();
+ modularLoadManagerAdmin.close();
+ modularLoadManager.close();
+ mockPulsarSetup1.cleanup();
+ mockPulsarSetup2.cleanup();
+ }
+
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java
index 5dc294d0d..3d3a1c23c 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java
@@ -563,7 +563,7 @@ void brokerStats() throws Exception {
doReturn("client-id").when(brokerStats).clientAppId();
Collection<Metrics> metrics = brokerStats.getMetrics();
assertNotNull(metrics);
- LoadReport loadReport = brokerStats.getLoadReport();
+ LoadReport loadReport = (LoadReport) brokerStats.getLoadReport();
assertNotNull(loadReport);
assertEquals(loadReport.isOverLoaded(), false);
Collection<Metrics> mBeans = brokerStats.getMBeans();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/BrokerVersionFilterTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/BrokerVersionFilterTest.java
index 4131d271a..760ef0732 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/BrokerVersionFilterTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/BrokerVersionFilterTest.java
@@ -22,11 +22,11 @@
import java.util.TreeSet;
import org.apache.pulsar.broker.BrokerData;
-import org.apache.pulsar.broker.LocalBrokerData;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.loadbalance.BrokerFilterBadVersionException;
import org.apache.pulsar.broker.loadbalance.LoadData;
import org.apache.pulsar.broker.loadbalance.impl.BrokerVersionFilter;
+import org.apache.pulsar.policies.data.loadbalancer.LocalBrokerData;
import org.testng.Assert;
import org.testng.annotations.Test;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/LoadBalancerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/LoadBalancerTest.java
index b99fad26b..40641187f 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/LoadBalancerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/LoadBalancerTest.java
@@ -101,8 +101,6 @@
public class LoadBalancerTest {
LocalBookkeeperEnsemble bkEnsemble;
- private static final ObjectMapper objectMapper = new ObjectMapper();
-
ExecutorService executor = new ThreadPoolExecutor(5, 20, 30, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());
private static final Logger log = LoggerFactory.getLogger(LoadBalancerTest.class);
@@ -202,7 +200,7 @@ public void testLoadReportsWrittenOnZK() throws Exception {
assert (loadReportData.length > 0);
log.info("LoadReport {}, {}", lookupAddresses[i], new String(loadReportData));
- LoadReport loadReport = objectMapper.readValue(loadReportData, LoadReport.class);
+ LoadReport loadReport = ObjectMapperFactory.getThreadLocal().readValue(loadReportData, LoadReport.class);
assert (loadReport.getName().equals(lookupAddresses[i]));
// Check Initial Ranking is populated in both the brokers
@@ -247,7 +245,7 @@ public void testUpdateLoadReportAndCheckUpdatedRanking() throws Exception {
String znodePath = String.format("%s/%s", SimpleLoadManagerImpl.LOADBALANCE_BROKERS_ROOT,
lookupAddresses[i]);
- String loadReportJson = objectMapper.writeValueAsString(lr);
+ String loadReportJson = ObjectMapperFactory.getThreadLocal().writeValueAsString(lr);
bkEnsemble.getZkClient().setData(znodePath, loadReportJson.getBytes(Charsets.UTF_8), -1);
}
@@ -318,7 +316,7 @@ public void testBrokerRanking() throws Exception {
String znodePath = String.format("%s/%s", SimpleLoadManagerImpl.LOADBALANCE_BROKERS_ROOT,
lookupAddresses[i]);
- String loadReportJson = objectMapper.writeValueAsString(lr);
+ String loadReportJson = ObjectMapperFactory.getThreadLocal().writeValueAsString(lr);
bkEnsemble.getZkClient().setData(znodePath, loadReportJson.getBytes(Charsets.UTF_8), -1);
}
@@ -379,7 +377,7 @@ public void testDestinationAssignmentWithExistingBundles() throws Exception {
String znodePath = String.format("%s/%s", SimpleLoadManagerImpl.LOADBALANCE_BROKERS_ROOT,
lookupAddresses[i]);
- String loadReportJson = objectMapper.writeValueAsString(lr);
+ String loadReportJson = ObjectMapperFactory.getThreadLocal().writeValueAsString(lr);
bkEnsemble.getZkClient().setData(znodePath, loadReportJson.getBytes(Charsets.UTF_8), -1);
}
@@ -454,7 +452,7 @@ public void testStop() throws Exception {
private void printResourceQuotas(Map<String, ResourceQuota> resourceQuotas) throws Exception {
log.info("Realtime Resource Quota:");
for (Map.Entry<String, ResourceQuota> entry : resourceQuotas.entrySet()) {
- String quotaStr = objectMapper.writeValueAsString(entry.getValue());
+ String quotaStr = ObjectMapperFactory.getThreadLocal().writeValueAsString(entry.getValue());
log.info(" {}, {}", entry.getKey(), quotaStr);
}
}
@@ -488,7 +486,7 @@ private void writeLoadReportsForDynamicQuota(long timestamp) throws Exception {
String znodePath = String.format("%s/%s", SimpleLoadManagerImpl.LOADBALANCE_BROKERS_ROOT,
lookupAddresses[i]);
- String loadReportJson = objectMapper.writeValueAsString(lr);
+ String loadReportJson = ObjectMapperFactory.getThreadLocal().writeValueAsString(lr);
bkEnsemble.getZkClient().setData(znodePath, loadReportJson.getBytes(Charsets.UTF_8), -1);
}
}
@@ -660,7 +658,7 @@ public void testNamespaceBundleAutoSplit() throws Exception {
setObjectField(SimpleLoadManagerImpl.class, pulsarServices[0].getLoadManager().get(), "lastLoadReport", lr);
String znodePath = String.format("%s/%s", SimpleLoadManagerImpl.LOADBALANCE_BROKERS_ROOT, lookupAddresses[0]);
- String loadReportJson = objectMapper.writeValueAsString(lr);
+ String loadReportJson = ObjectMapperFactory.getThreadLocal().writeValueAsString(lr);
bkEnsemble.getZkClient().setData(znodePath, loadReportJson.getBytes(Charsets.UTF_8), -1);
// sleep to wait load ranking be triggered and trigger bundle split
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/ModularLoadManagerImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/ModularLoadManagerImplTest.java
index 1bd51387a..c17b78db2 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/ModularLoadManagerImplTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/ModularLoadManagerImplTest.java
@@ -44,7 +44,6 @@
import org.apache.bookkeeper.util.ZkUtils;
import org.apache.pulsar.broker.BrokerData;
import org.apache.pulsar.broker.BundleData;
-import org.apache.pulsar.broker.LocalBrokerData;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.TimeAverageMessageData;
@@ -59,6 +58,7 @@
import org.apache.pulsar.common.naming.NamespaceBundles;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.ServiceUnitId;
+import org.apache.pulsar.policies.data.loadbalancer.LocalBrokerData;
import org.apache.pulsar.policies.data.loadbalancer.NamespaceBundleStats;
import org.apache.pulsar.policies.data.loadbalancer.ResourceUsage;
import org.apache.pulsar.policies.data.loadbalancer.SystemResourceUsage;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/ModularLoadManagerStrategyTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/ModularLoadManagerStrategyTest.java
index 78ae4489b..d06fb087f 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/ModularLoadManagerStrategyTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/ModularLoadManagerStrategyTest.java
@@ -22,12 +22,12 @@
import org.apache.pulsar.broker.BrokerData;
import org.apache.pulsar.broker.BundleData;
-import org.apache.pulsar.broker.LocalBrokerData;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.TimeAverageBrokerData;
import org.apache.pulsar.broker.loadbalance.LoadData;
import org.apache.pulsar.broker.loadbalance.ModularLoadManagerStrategy;
import org.apache.pulsar.broker.loadbalance.impl.LeastLongTermMessageRate;
+import org.apache.pulsar.policies.data.loadbalancer.LocalBrokerData;
import org.apache.pulsar.policies.data.loadbalancer.ResourceUsage;
import org.testng.annotations.Test;
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 560547d3f..aa0979b04 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
@@ -44,7 +44,6 @@
import org.apache.bookkeeper.util.ZkUtils;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.tuple.Pair;
-import org.apache.pulsar.broker.LocalBrokerData;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.cache.LocalZooKeeperCacheService;
import org.apache.pulsar.broker.loadbalance.LoadManager;
@@ -66,6 +65,7 @@
import org.apache.pulsar.common.util.ObjectMapperFactory;
import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
import org.apache.pulsar.policies.data.loadbalancer.LoadReport;
+import org.apache.pulsar.policies.data.loadbalancer.LocalBrokerData;
import org.apache.pulsar.zookeeper.ZooKeeperDataCache;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/BrokerStats.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/BrokerStats.java
index 68a25cf51..6fe07c70a 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/BrokerStats.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/BrokerStats.java
@@ -19,6 +19,7 @@
package org.apache.pulsar.client.admin;
import org.apache.pulsar.common.stats.AllocatorStats;
+import org.apache.pulsar.policies.data.loadbalancer.LoadManagerReport;
import com.google.gson.JsonArray;
import com.google.gson.JsonObject;
@@ -60,4 +61,6 @@
JsonObject getPendingBookieOpsStats() throws PulsarAdminException;
AllocatorStats getAllocatorStats(String allocatorName) throws PulsarAdminException;
+
+ LoadManagerReport getLoadReport() throws PulsarAdminException;
}
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BrokerStatsImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BrokerStatsImpl.java
index aecff5b74..917fd6571 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BrokerStatsImpl.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BrokerStatsImpl.java
@@ -24,6 +24,8 @@
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.api.Authentication;
import org.apache.pulsar.common.stats.AllocatorStats;
+import org.apache.pulsar.policies.data.loadbalancer.LoadManagerReport;
+import org.apache.pulsar.policies.data.loadbalancer.LocalBrokerData;
import com.google.gson.Gson;
import com.google.gson.JsonArray;
@@ -82,10 +84,10 @@ public JsonObject getDestinations() throws PulsarAdminException {
}
}
- public JsonObject getLoadReport() throws PulsarAdminException {
+ @Override
+ public LoadManagerReport getLoadReport() throws PulsarAdminException {
try {
- String json = request(brokerStats.path("/load-report")).get(String.class);
- return new Gson().fromJson(json, JsonObject.class);
+ return request(brokerStats.path("/load-report")).get(LocalBrokerData.class);
} catch (Exception e) {
throw getApiException(e);
}
diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdBrokerStats.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdBrokerStats.java
index f0ce39311..0bea28b2b 100644
--- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdBrokerStats.java
+++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdBrokerStats.java
@@ -26,6 +26,7 @@
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.common.stats.AllocatorStats;
import org.apache.pulsar.common.util.ObjectMapperFactory;
+import org.apache.pulsar.policies.data.loadbalancer.LoadManagerReport;
import com.google.gson.Gson;
import com.google.gson.JsonArray;
@@ -86,6 +87,15 @@ void run() throws Exception {
}
+ @Parameters(commandDescription = "dump broker load-report")
+ public class CmdLoadReport extends CliCommand {
+
+ @Override
+ void run() throws Exception {
+ print(admin.brokerStats().getLoadReport());
+ }
+ }
+
@Parameters(commandDescription = "dump destination stats")
public class CmdDestinations extends CliCommand {
@Parameter(names = { "-i", "--indent" }, description = "Indent JSON output", required = false)
@@ -126,6 +136,7 @@ public CmdBrokerStats(PulsarAdmin admin) {
jcommander.addCommand("mbeans", new CmdDumpMBeans());
jcommander.addCommand("destinations", new CmdDestinations());
jcommander.addCommand("allocator-stats", new CmdAllocatorStats());
+ jcommander.addCommand("load-report", new CmdLoadReport());
}
public void addCommands(String commandName, CliCommand cliCommand) {
diff --git a/pulsar-client-tools/src/test/java/org/apache/pulsar/admin/cli/PulsarAdminToolTest.java b/pulsar-client-tools/src/test/java/org/apache/pulsar/admin/cli/PulsarAdminToolTest.java
index 7725d2b91..306579eea 100644
--- a/pulsar-client-tools/src/test/java/org/apache/pulsar/admin/cli/PulsarAdminToolTest.java
+++ b/pulsar-client-tools/src/test/java/org/apache/pulsar/admin/cli/PulsarAdminToolTest.java
@@ -26,6 +26,7 @@
import java.util.EnumSet;
+import org.apache.pulsar.client.admin.BrokerStats;
import org.apache.pulsar.client.admin.Brokers;
import org.apache.pulsar.client.admin.Clusters;
import org.apache.pulsar.client.admin.Lookup;
@@ -75,6 +76,27 @@ void brokers() throws Exception {
}
@Test
+ void brokerStats() throws Exception {
+ PulsarAdmin admin = Mockito.mock(PulsarAdmin.class);
+ BrokerStats mockBrokerStats = mock(BrokerStats.class);
+ doReturn(mockBrokerStats).when(admin).brokerStats();
+
+ CmdBrokerStats brokerStats = new CmdBrokerStats(admin);
+
+ brokerStats.run(split("destinations"));
+ verify(mockBrokerStats).getDestinations();
+
+ brokerStats.run(split("load-report"));
+ verify(mockBrokerStats).getLoadReport();
+
+ brokerStats.run(split("mbeans"));
+ verify(mockBrokerStats).getMBeans();
+
+ brokerStats.run(split("monitoring-metrics"));
+ verify(mockBrokerStats).getMetrics();
+ }
+
+ @Test
void getOwnedNamespaces() throws Exception {
PulsarAdmin admin = Mockito.mock(PulsarAdmin.class);
Brokers mockBrokers = mock(Brokers.class);
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/policies/data/loadbalancer/NamespaceBundleStatsComparator.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/NamespaceBundleStatsComparator.java
similarity index 91%
rename from pulsar-common/src/main/java/org/apache/pulsar/policies/data/loadbalancer/NamespaceBundleStatsComparator.java
rename to pulsar-common/src/main/java/org/apache/pulsar/common/util/NamespaceBundleStatsComparator.java
index f311ada52..bbab28e12 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/policies/data/loadbalancer/NamespaceBundleStatsComparator.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/NamespaceBundleStatsComparator.java
@@ -16,11 +16,13 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.pulsar.policies.data.loadbalancer;
+package org.apache.pulsar.common.util;
import java.util.Comparator;
import java.util.Map;
+import org.apache.pulsar.policies.data.loadbalancer.NamespaceBundleStats;
+import org.apache.pulsar.policies.data.loadbalancer.SystemResourceUsage;
import org.apache.pulsar.policies.data.loadbalancer.SystemResourceUsage.ResourceType;
/**
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/JSONWritable.java b/pulsar-common/src/main/java/org/apache/pulsar/policies/data/loadbalancer/JSONWritable.java
similarity index 96%
rename from pulsar-broker/src/main/java/org/apache/pulsar/broker/JSONWritable.java
rename to pulsar-common/src/main/java/org/apache/pulsar/policies/data/loadbalancer/JSONWritable.java
index 430dd572b..f197ab6b7 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/JSONWritable.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/policies/data/loadbalancer/JSONWritable.java
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.pulsar.broker;
+package org.apache.pulsar.policies.data.loadbalancer;
import org.apache.pulsar.common.util.ObjectMapperFactory;
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/policies/data/loadbalancer/LoadManagerReport.java b/pulsar-common/src/main/java/org/apache/pulsar/policies/data/loadbalancer/LoadManagerReport.java
new file mode 100644
index 000000000..cabca25bc
--- /dev/null
+++ b/pulsar-common/src/main/java/org/apache/pulsar/policies/data/loadbalancer/LoadManagerReport.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.policies.data.loadbalancer;
+
+import java.util.Map;
+
+/**
+ * This class represents the overall load of the broker - it includes overall SystemResourceUsage and Bundle-usage
+ */
+public interface LoadManagerReport extends ServiceLookupData {
+
+ public ResourceUsage getCpu();
+
+ public ResourceUsage getMemory();
+
+ public ResourceUsage getDirectMemory();
+
+ public ResourceUsage getBandwidthIn();
+
+ public ResourceUsage getBandwidthOut();
+
+ public long getLastUpdate();
+
+ public Map<String, NamespaceBundleStats> getBundleStats();
+
+ public int getNumTopics();
+
+ public int getNumBundles();
+
+ public int getNumConsumers();
+
+ public int getNumProducers();
+
+ public double getMsgThroughputIn();
+
+ public double getMsgThroughputOut();
+
+ public double getMsgRateIn();
+
+ public double getMsgRateOut();
+
+ public String getBrokerVersionString();
+
+ public boolean isPersistentTopicsEnabled();
+
+ public boolean isNonPersistentTopicsEnabled();
+
+}
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/policies/data/loadbalancer/LoadReport.java b/pulsar-common/src/main/java/org/apache/pulsar/policies/data/loadbalancer/LoadReport.java
index 474f76970..2ea41a14a 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/policies/data/loadbalancer/LoadReport.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/policies/data/loadbalancer/LoadReport.java
@@ -24,17 +24,17 @@
import java.util.Set;
import java.util.TreeMap;
+import org.apache.pulsar.common.util.NamespaceBundleStatsComparator;
import org.apache.pulsar.policies.data.loadbalancer.SystemResourceUsage.ResourceType;
import com.fasterxml.jackson.annotation.JsonIgnore;
import com.google.common.collect.Maps;
-
/**
* This class represents the overall load of the broker - it includes overall {@link SystemResourceUsage} and
* {@link NamespaceUsage} for all the namespaces hosted by this broker.
*/
-public class LoadReport implements ServiceLookupData {
+public class LoadReport implements LoadManagerReport {
private String name;
private String brokerVersionString;
@@ -42,24 +42,25 @@
private final String webServiceUrlTls;
private final String pulsarServiceUrl;
private final String pulsarServiceUrlTls;
- private boolean persistentTopicsEnabled=true;
- private boolean nonPersistentTopicsEnabled=true;
-
+ private boolean persistentTopicsEnabled = true;
+ private boolean nonPersistentTopicsEnabled = true;
+
private boolean isUnderLoaded;
private boolean isOverLoaded;
private long timestamp;
private double msgRateIn;
private double msgRateOut;
- private long numTopics;
- private long numConsumers;
- private long numProducers;
- private long numBundles;
+ private int numTopics;
+ private int numConsumers;
+ private int numProducers;
+ private int numBundles;
public LoadReport() {
- this(null, null, null, null);
+ this(null, null, null, null);
}
- public LoadReport(String webServiceUrl, String webServiceUrlTls, String pulsarServiceUrl, String pulsarServiceUrlTls) {
+ public LoadReport(String webServiceUrl, String webServiceUrlTls, String pulsarServiceUrl,
+ String pulsarServiceUrlTls) {
this.webServiceUrl = webServiceUrl;
this.webServiceUrlTls = webServiceUrlTls;
this.pulsarServiceUrl = pulsarServiceUrl;
@@ -192,7 +193,8 @@ public double getMsgRateOut() {
return msgRateOut;
}
- public long getNumTopics() {
+ @Override
+ public int getNumTopics() {
numTopics = 0;
if (this.bundleStats != null) {
this.bundleStats.forEach((bundle, stats) -> {
@@ -202,7 +204,8 @@ public long getNumTopics() {
return numTopics;
}
- public long getNumConsumers() {
+ @Override
+ public int getNumConsumers() {
numConsumers = 0;
if (this.bundleStats != null) {
for (Map.Entry<String, NamespaceBundleStats> entry : this.bundleStats.entrySet()) {
@@ -212,7 +215,8 @@ public long getNumConsumers() {
return numConsumers;
}
- public long getNumProducers() {
+ @Override
+ public int getNumProducers() {
numProducers = 0;
if (this.bundleStats != null) {
for (Map.Entry<String, NamespaceBundleStats> entry : this.bundleStats.entrySet()) {
@@ -222,7 +226,8 @@ public long getNumProducers() {
return numProducers;
}
- public long getNumBundles() {
+ @Override
+ public int getNumBundles() {
numBundles = 0;
if (this.bundleStats != null) {
numBundles = this.bundleStats.size();
@@ -406,4 +411,44 @@ public boolean isNonPersistentTopicsEnabled() {
public void setNonPersistentTopicsEnabled(boolean nonPersistentTopicsEnabled) {
this.nonPersistentTopicsEnabled = nonPersistentTopicsEnabled;
}
+
+ @Override
+ public ResourceUsage getCpu() {
+ return systemResourceUsage != null ? systemResourceUsage.cpu : null;
+ }
+
+ @Override
+ public ResourceUsage getMemory() {
+ return systemResourceUsage != null ? systemResourceUsage.memory : null;
+ }
+
+ @Override
+ public ResourceUsage getDirectMemory() {
+ return systemResourceUsage != null ? systemResourceUsage.directMemory : null;
+ }
+
+ @Override
+ public ResourceUsage getBandwidthIn() {
+ return systemResourceUsage != null ? systemResourceUsage.bandwidthIn : null;
+ }
+
+ @Override
+ public ResourceUsage getBandwidthOut() {
+ return systemResourceUsage != null ? systemResourceUsage.bandwidthOut : null;
+ }
+
+ @Override
+ public long getLastUpdate() {
+ return timestamp;
+ }
+
+ @Override
+ public double getMsgThroughputIn() {
+ return msgRateIn;
+ }
+
+ @Override
+ public double getMsgThroughputOut() {
+ return msgRateOut;
+ }
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/LocalBrokerData.java b/pulsar-common/src/main/java/org/apache/pulsar/policies/data/loadbalancer/LocalBrokerData.java
similarity index 96%
rename from pulsar-broker/src/main/java/org/apache/pulsar/broker/LocalBrokerData.java
rename to pulsar-common/src/main/java/org/apache/pulsar/policies/data/loadbalancer/LocalBrokerData.java
index 6184c2446..a645e6cf8 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/LocalBrokerData.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/policies/data/loadbalancer/LocalBrokerData.java
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.pulsar.broker;
+package org.apache.pulsar.policies.data.loadbalancer;
import java.util.HashMap;
import java.util.HashSet;
@@ -24,15 +24,11 @@
import java.util.Map;
import java.util.Set;
-import org.apache.pulsar.policies.data.loadbalancer.NamespaceBundleStats;
-import org.apache.pulsar.policies.data.loadbalancer.ResourceUsage;
-import org.apache.pulsar.policies.data.loadbalancer.ServiceLookupData;
-import org.apache.pulsar.policies.data.loadbalancer.SystemResourceUsage;
/**
* Contains all the data that is maintained locally on each broker.
*/
-public class LocalBrokerData extends JSONWritable implements ServiceLookupData {
+public class LocalBrokerData extends JSONWritable implements LoadManagerReport {
// URLs to satisfy contract of ServiceLookupData (used by NamespaceService).
private final String webServiceUrl;
@@ -201,6 +197,7 @@ public double getMaxResourceUsage() {
/ 100;
}
+ @Override
public ResourceUsage getCpu() {
return cpu;
}
@@ -209,6 +206,7 @@ public void setCpu(ResourceUsage cpu) {
this.cpu = cpu;
}
+ @Override
public ResourceUsage getMemory() {
return memory;
}
@@ -217,6 +215,7 @@ public void setMemory(ResourceUsage memory) {
this.memory = memory;
}
+ @Override
public ResourceUsage getDirectMemory() {
return directMemory;
}
@@ -225,6 +224,7 @@ public void setDirectMemory(ResourceUsage directMemory) {
this.directMemory = directMemory;
}
+ @Override
public ResourceUsage getBandwidthIn() {
return bandwidthIn;
}
@@ -233,6 +233,7 @@ public void setBandwidthIn(ResourceUsage bandwidthIn) {
this.bandwidthIn = bandwidthIn;
}
+ @Override
public ResourceUsage getBandwidthOut() {
return bandwidthOut;
}
@@ -257,6 +258,7 @@ public void setLastBundleLosses(Set<String> lastBundleLosses) {
this.lastBundleLosses = lastBundleLosses;
}
+ @Override
public long getLastUpdate() {
return lastUpdate;
}
@@ -281,6 +283,7 @@ public void setLastStats(Map<String, NamespaceBundleStats> lastStats) {
this.lastStats = lastStats;
}
+ @Override
public int getNumTopics() {
return numTopics;
}
@@ -289,6 +292,7 @@ public void setNumTopics(int numTopics) {
this.numTopics = numTopics;
}
+ @Override
public int getNumBundles() {
return numBundles;
}
@@ -297,6 +301,7 @@ public void setNumBundles(int numBundles) {
this.numBundles = numBundles;
}
+ @Override
public int getNumConsumers() {
return numConsumers;
}
@@ -305,6 +310,7 @@ public void setNumConsumers(int numConsumers) {
this.numConsumers = numConsumers;
}
+ @Override
public int getNumProducers() {
return numProducers;
}
@@ -313,6 +319,7 @@ public void setNumProducers(int numProducers) {
this.numProducers = numProducers;
}
+ @Override
public double getMsgThroughputIn() {
return msgThroughputIn;
}
@@ -321,6 +328,7 @@ public void setMsgThroughputIn(double msgThroughputIn) {
this.msgThroughputIn = msgThroughputIn;
}
+ @Override
public double getMsgThroughputOut() {
return msgThroughputOut;
}
@@ -329,6 +337,7 @@ public void setMsgThroughputOut(double msgThroughputOut) {
this.msgThroughputOut = msgThroughputOut;
}
+ @Override
public double getMsgRateIn() {
return msgRateIn;
}
@@ -337,6 +346,7 @@ public void setMsgRateIn(double msgRateIn) {
this.msgRateIn = msgRateIn;
}
+ @Override
public double getMsgRateOut() {
return msgRateOut;
}
@@ -349,6 +359,7 @@ public void setBrokerVersionString(String brokerVersionString) {
this.brokerVersionString = brokerVersionString;
}
+ @Override
public String getBrokerVersionString() {
return brokerVersionString;
}
@@ -373,6 +384,7 @@ public String getPulsarServiceUrlTls() {
return pulsarServiceUrlTls;
}
+ @Override
public boolean isPersistentTopicsEnabled() {
return persistentTopicsEnabled;
}
@@ -381,6 +393,7 @@ public void setPersistentTopicsEnabled(boolean persistentTopicsEnabled) {
this.persistentTopicsEnabled = persistentTopicsEnabled;
}
+ @Override
public boolean isNonPersistentTopicsEnabled() {
return nonPersistentTopicsEnabled;
}
@@ -389,4 +402,9 @@ public void setNonPersistentTopicsEnabled(boolean nonPersistentTopicsEnabled) {
this.nonPersistentTopicsEnabled = nonPersistentTopicsEnabled;
}
+ @Override
+ public Map<String, NamespaceBundleStats> getBundleStats() {
+ return getLastStats();
+ }
+
}
diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/BrokerMonitor.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/BrokerMonitor.java
index 1dea71ed0..6978e30c1 100644
--- a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/BrokerMonitor.java
+++ b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/BrokerMonitor.java
@@ -27,10 +27,10 @@
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
-import org.apache.pulsar.broker.LocalBrokerData;
import org.apache.pulsar.broker.TimeAverageBrokerData;
import org.apache.pulsar.broker.loadbalance.impl.ModularLoadManagerImpl;
import org.apache.pulsar.policies.data.loadbalancer.LoadReport;
+import org.apache.pulsar.policies.data.loadbalancer.LocalBrokerData;
import org.apache.pulsar.policies.data.loadbalancer.ResourceUsage;
import org.apache.pulsar.policies.data.loadbalancer.SystemResourceUsage;
import org.apache.pulsar.testclient.utils.FixedColumnLengthTableMaker;
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services