You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by zi...@apache.org on 2024/05/10 15:35:27 UTC
(pulsar) branch branch-3.2 updated: [fix][broker] usedLocallySinceLastReport should always be reset (#22672)
This is an automated email from the ASF dual-hosted git repository.
zixuan pushed a commit to branch branch-3.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-3.2 by this push:
new 1f0afdd6738 [fix][broker] usedLocallySinceLastReport should always be reset (#22672)
1f0afdd6738 is described below
commit 1f0afdd6738b935d11a4004e589487c66c48cb5a
Author: Zixuan Liu <no...@gmail.com>
AuthorDate: Thu May 9 09:42:17 2024 +0800
[fix][broker] usedLocallySinceLastReport should always be reset (#22672)
Signed-off-by: Zixuan Liu <no...@gmail.com>
(cherry picked from commit 8f015d89e5d246325ae5cada02c4af3017a97ed9)
---
.../pulsar/broker/resourcegroup/ResourceGroup.java | 3 +-
.../ResourceGroupReportLocalUsageTest.java | 50 ++++++++++++++--------
2 files changed, 34 insertions(+), 19 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceGroup.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceGroup.java
index b4aacf0b1f0..c3ebaacb0da 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceGroup.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceGroup.java
@@ -456,14 +456,13 @@ public class ResourceGroup {
bytesUsed = monEntity.usedLocallySinceLastReport.bytes;
messagesUsed = monEntity.usedLocallySinceLastReport.messages;
-
+ monEntity.usedLocallySinceLastReport.bytes = monEntity.usedLocallySinceLastReport.messages = 0;
if (sendReport) {
p.setBytesPerPeriod(bytesUsed);
p.setMessagesPerPeriod(messagesUsed);
monEntity.lastReportedValues.bytes = bytesUsed;
monEntity.lastReportedValues.messages = messagesUsed;
monEntity.numSuppressedUsageReports = 0;
- monEntity.usedLocallySinceLastReport.bytes = monEntity.usedLocallySinceLastReport.messages = 0;
monEntity.totalUsedLocally.bytes += bytesUsed;
monEntity.totalUsedLocally.messages += messagesUsed;
monEntity.lastResourceUsageFillTimeMSecsSinceEpoch = System.currentTimeMillis();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupReportLocalUsageTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupReportLocalUsageTest.java
index 658b7c94165..139d19886c7 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupReportLocalUsageTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupReportLocalUsageTest.java
@@ -72,34 +72,50 @@ public class ResourceGroupReportLocalUsageTest extends MockedPulsarServiceBaseTe
rgConfig.setPublishRateInMsgs(2000);
service.resourceGroupCreate(rgName, rgConfig);
- org.apache.pulsar.broker.resourcegroup.ResourceGroup resourceGroup = service.resourceGroupGet(rgName);
BytesAndMessagesCount bytesAndMessagesCount = new BytesAndMessagesCount();
bytesAndMessagesCount.bytes = 20;
bytesAndMessagesCount.messages = 10;
- resourceGroup.incrementLocalUsageStats(ResourceGroupMonitoringClass.Publish, bytesAndMessagesCount);
+
+ org.apache.pulsar.broker.resourcegroup.ResourceGroup resourceGroup = service.resourceGroupGet(rgName);
+ for (ResourceGroupMonitoringClass value : ResourceGroupMonitoringClass.values()) {
+ resourceGroup.incrementLocalUsageStats(value, bytesAndMessagesCount);
+ }
+
+ // Case1: Suppress report ResourceUsage.
+ needReport.set(false);
ResourceUsage resourceUsage = new ResourceUsage();
resourceGroup.rgFillResourceUsage(resourceUsage);
assertFalse(resourceUsage.hasDispatch());
assertFalse(resourceUsage.hasPublish());
+ for (ResourceGroupMonitoringClass value : ResourceGroupMonitoringClass.values()) {
+ PerMonitoringClassFields monitoredEntity =
+ resourceGroup.getMonitoredEntity(value);
+ assertEquals(monitoredEntity.usedLocallySinceLastReport.messages, 0);
+ assertEquals(monitoredEntity.usedLocallySinceLastReport.bytes, 0);
+ assertEquals(monitoredEntity.totalUsedLocally.messages, 0);
+ assertEquals(monitoredEntity.totalUsedLocally.bytes, 0);
+ assertEquals(monitoredEntity.lastReportedValues.messages, 0);
+ assertEquals(monitoredEntity.lastReportedValues.bytes, 0);
+ }
- PerMonitoringClassFields publishMonitoredEntity =
- resourceGroup.getMonitoredEntity(ResourceGroupMonitoringClass.Publish);
- assertEquals(publishMonitoredEntity.usedLocallySinceLastReport.messages, bytesAndMessagesCount.messages);
- assertEquals(publishMonitoredEntity.usedLocallySinceLastReport.bytes, bytesAndMessagesCount.bytes);
- assertEquals(publishMonitoredEntity.totalUsedLocally.messages, 0);
- assertEquals(publishMonitoredEntity.totalUsedLocally.bytes, 0);
- assertEquals(publishMonitoredEntity.lastReportedValues.messages, 0);
- assertEquals(publishMonitoredEntity.lastReportedValues.bytes, 0);
-
+ // Case2: Report ResourceUsage.
+ for (ResourceGroupMonitoringClass value : ResourceGroupMonitoringClass.values()) {
+ resourceGroup.incrementLocalUsageStats(value, bytesAndMessagesCount);
+ }
needReport.set(true);
+ resourceUsage = new ResourceUsage();
resourceGroup.rgFillResourceUsage(resourceUsage);
assertTrue(resourceUsage.hasDispatch());
assertTrue(resourceUsage.hasPublish());
- assertEquals(publishMonitoredEntity.usedLocallySinceLastReport.messages, 0);
- assertEquals(publishMonitoredEntity.usedLocallySinceLastReport.bytes, 0);
- assertEquals(publishMonitoredEntity.totalUsedLocally.messages, bytesAndMessagesCount.messages);
- assertEquals(publishMonitoredEntity.totalUsedLocally.bytes, bytesAndMessagesCount.bytes);
- assertEquals(publishMonitoredEntity.lastReportedValues.messages, bytesAndMessagesCount.messages);
- assertEquals(publishMonitoredEntity.lastReportedValues.bytes, bytesAndMessagesCount.bytes);
+ for (ResourceGroupMonitoringClass value : ResourceGroupMonitoringClass.values()) {
+ PerMonitoringClassFields monitoredEntity =
+ resourceGroup.getMonitoredEntity(value);
+ assertEquals(monitoredEntity.usedLocallySinceLastReport.messages, 0);
+ assertEquals(monitoredEntity.usedLocallySinceLastReport.bytes, 0);
+ assertEquals(monitoredEntity.totalUsedLocally.messages, bytesAndMessagesCount.messages);
+ assertEquals(monitoredEntity.totalUsedLocally.bytes, bytesAndMessagesCount.bytes);
+ assertEquals(monitoredEntity.lastReportedValues.messages, bytesAndMessagesCount.messages);
+ assertEquals(monitoredEntity.lastReportedValues.bytes, bytesAndMessagesCount.bytes);
+ }
}
}
\ No newline at end of file