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/03/26 09:29:52 UTC
(pulsar) branch branch-3.2 updated (5c3c1f115ea -> 97afd308a57)
This is an automated email from the ASF dual-hosted git repository.
zixuan pushed a change to branch branch-3.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 5c3c1f115ea [fix] [broker] fix mismatch between dispatcher.consumerList and dispatcher.consumerSet (#22283)
new 36ba4766f61 [fix][broker] Fix ResourceGroup report local usage (#22340)
new 97afd308a57 [fix][broker] Fix ResourceGroups loading (#21781)
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../pulsar/broker/resourcegroup/ResourceGroup.java | 38 +++++---
.../resourcegroup/ResourceGroupConfigListener.java | 76 +++++++++++----
.../broker/resourcegroup/ResourceGroupService.java | 4 +-
.../ResourceGroupConfigListenerTest.java | 48 ++++++++++
.../ResourceGroupReportLocalUsageTest.java | 105 +++++++++++++++++++++
5 files changed, 234 insertions(+), 37 deletions(-)
create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupReportLocalUsageTest.java
(pulsar) 02/02: [fix][broker] Fix ResourceGroups loading (#21781)
Posted by zi...@apache.org.
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
commit 97afd308a5759ac47125ae5599b4de7c6bba8bb4
Author: Zixuan Liu <no...@gmail.com>
AuthorDate: Tue Mar 26 16:49:58 2024 +0800
[fix][broker] Fix ResourceGroups loading (#21781)
Signed-off-by: Zixuan Liu <no...@gmail.com>
(cherry picked from commit 80b491dab0fd8a948db0a3d85a3ccb8490ecf266)
---
.../resourcegroup/ResourceGroupConfigListener.java | 76 ++++++++++++++++------
.../ResourceGroupConfigListenerTest.java | 48 ++++++++++++++
2 files changed, 104 insertions(+), 20 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupConfigListener.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupConfigListener.java
index c15edd2be4e..4a5b8a8bcc2 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupConfigListener.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupConfigListener.java
@@ -18,15 +18,22 @@
*/
package org.apache.pulsar.broker.resourcegroup;
+import static org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Sets;
+import java.util.ArrayList;
import java.util.HashSet;
+import java.util.List;
import java.util.Optional;
import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.resources.ResourceGroupResources;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.common.policies.data.ResourceGroup;
+import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.metadata.api.Notification;
import org.apache.pulsar.metadata.api.NotificationType;
import org.slf4j.Logger;
@@ -47,24 +54,32 @@ public class ResourceGroupConfigListener implements Consumer<Notification> {
private final ResourceGroupService rgService;
private final PulsarService pulsarService;
private final ResourceGroupResources rgResources;
- private final ResourceGroupNamespaceConfigListener rgNamespaceConfigListener;
+ private volatile ResourceGroupNamespaceConfigListener rgNamespaceConfigListener;
public ResourceGroupConfigListener(ResourceGroupService rgService, PulsarService pulsarService) {
this.rgService = rgService;
this.pulsarService = pulsarService;
this.rgResources = pulsarService.getPulsarResources().getResourcegroupResources();
- loadAllResourceGroups();
this.rgResources.getStore().registerListener(this);
- rgNamespaceConfigListener = new ResourceGroupNamespaceConfigListener(
- rgService, pulsarService, this);
+ execute(() -> loadAllResourceGroupsWithRetryAsync(0));
}
- private void loadAllResourceGroups() {
- rgResources.listResourceGroupsAsync().whenCompleteAsync((rgList, ex) -> {
- if (ex != null) {
- LOG.error("Exception when fetching resource groups", ex);
- return;
+ private void loadAllResourceGroupsWithRetryAsync(long retry) {
+ loadAllResourceGroupsAsync().thenAccept(__ -> {
+ if (rgNamespaceConfigListener == null) {
+ rgNamespaceConfigListener = new ResourceGroupNamespaceConfigListener(rgService, pulsarService, this);
}
+ }).exceptionally(e -> {
+ long nextRetry = retry + 1;
+ long delay = 500 * nextRetry;
+ LOG.error("Failed to load all resource groups during initialization, retrying after {}ms: ", delay, e);
+ schedule(() -> loadAllResourceGroupsWithRetryAsync(nextRetry), delay);
+ return null;
+ });
+ }
+
+ private CompletableFuture<Void> loadAllResourceGroupsAsync() {
+ return rgResources.listResourceGroupsAsync().thenCompose(rgList -> {
final Set<String> existingSet = rgService.resourceGroupGetAll();
HashSet<String> newSet = new HashSet<>();
@@ -72,21 +87,26 @@ public class ResourceGroupConfigListener implements Consumer<Notification> {
final Sets.SetView<String> deleteList = Sets.difference(existingSet, newSet);
- for (String rgName: deleteList) {
+ for (String rgName : deleteList) {
deleteResourceGroup(rgName);
}
final Sets.SetView<String> addList = Sets.difference(newSet, existingSet);
- for (String rgName: addList) {
- pulsarService.getPulsarResources().getResourcegroupResources()
- .getResourceGroupAsync(rgName).thenAcceptAsync(optionalRg -> {
- ResourceGroup rg = optionalRg.get();
- createResourceGroup(rgName, rg);
- }).exceptionally((ex1) -> {
- LOG.error("Failed to fetch resourceGroup", ex1);
- return null;
- });
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+ for (String rgName : addList) {
+ futures.add(pulsarService.getPulsarResources()
+ .getResourcegroupResources()
+ .getResourceGroupAsync(rgName)
+ .thenAccept(optionalRg -> {
+ if (optionalRg.isPresent()) {
+ ResourceGroup rg = optionalRg.get();
+ createResourceGroup(rgName, rg);
+ }
+ })
+ );
}
+
+ return FutureUtil.waitForAll(futures);
});
}
@@ -140,7 +160,10 @@ public class ResourceGroupConfigListener implements Consumer<Notification> {
Optional<String> rgName = ResourceGroupResources.resourceGroupNameFromPath(notifyPath);
if ((notification.getType() == NotificationType.ChildrenChanged)
|| (notification.getType() == NotificationType.Created)) {
- loadAllResourceGroups();
+ loadAllResourceGroupsAsync().exceptionally((ex) -> {
+ LOG.error("Exception when fetching resource groups", ex);
+ return null;
+ });
} else if (rgName.isPresent()) {
switch (notification.getType()) {
case Modified:
@@ -151,4 +174,17 @@ public class ResourceGroupConfigListener implements Consumer<Notification> {
}
}
}
+
+ protected void execute(Runnable runnable) {
+ pulsarService.getExecutor().execute(catchingAndLoggingThrowables(runnable));
+ }
+
+ protected void schedule(Runnable runnable, long delayMs) {
+ pulsarService.getExecutor().schedule(catchingAndLoggingThrowables(runnable), delayMs, TimeUnit.MILLISECONDS);
+ }
+
+ @VisibleForTesting
+ ResourceGroupNamespaceConfigListener getRgNamespaceConfigListener() {
+ return rgNamespaceConfigListener;
+ }
}
\ No newline at end of file
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupConfigListenerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupConfigListenerTest.java
index 90c26530850..4010635ed99 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupConfigListenerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupConfigListenerTest.java
@@ -18,20 +18,31 @@
*/
package org.apache.pulsar.broker.resourcegroup;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertThrows;
import com.google.common.collect.Sets;
+import java.util.ArrayList;
import java.util.Random;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
+import org.apache.pulsar.broker.resources.PulsarResources;
import org.apache.pulsar.broker.resources.ResourceGroupResources;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.policies.data.ClusterData;
import org.apache.pulsar.common.policies.data.ResourceGroup;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
+import org.apache.pulsar.metadata.api.MetadataStore;
import org.awaitility.Awaitility;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
@@ -288,4 +299,41 @@ public class ResourceGroupConfigListenerTest extends MockedPulsarServiceBaseTest
testAddRg.setDispatchRateInBytes(200L);
}
+
+ @Test
+ public void testNewResourceGroupNamespaceConfigListener() {
+ PulsarService pulsarService = mock(PulsarService.class);
+ PulsarResources pulsarResources = mock(PulsarResources.class);
+ doReturn(pulsarResources).when(pulsarService).getPulsarResources();
+ ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
+ doReturn(scheduledExecutorService).when(pulsarService).getExecutor();
+
+ ResourceGroupService resourceGroupService = mock(ResourceGroupService.class);
+ ResourceGroupResources resourceGroupResources = mock(ResourceGroupResources.class);
+ RuntimeException exception = new RuntimeException("listResourceGroupsAsync error");
+ doReturn(CompletableFuture.failedFuture(exception))
+ .when(resourceGroupResources).listResourceGroupsAsync();
+ doReturn(mock(MetadataStore.class))
+ .when(resourceGroupResources).getStore();
+ doReturn(resourceGroupResources).when(pulsarResources).getResourcegroupResources();
+
+ ServiceConfiguration ServiceConfiguration = new ServiceConfiguration();
+ doReturn(ServiceConfiguration).when(pulsarService).getConfiguration();
+
+ ResourceGroupConfigListener resourceGroupConfigListener =
+ new ResourceGroupConfigListener(resourceGroupService, pulsarService);
+
+ // getResourcegroupResources() returns an error, ResourceGroupNamespaceConfigListener doesn't be created.
+ Awaitility.await().pollDelay(3, TimeUnit.SECONDS).untilAsserted(() -> {
+ assertNull(resourceGroupConfigListener.getRgNamespaceConfigListener());
+ });
+
+ // ResourceGroupNamespaceConfigListener will be created, and uses real pulsar resource.
+ doReturn(CompletableFuture.completedFuture(new ArrayList<String>()))
+ .when(resourceGroupResources).listResourceGroupsAsync();
+ doReturn(pulsar.getPulsarResources()).when(pulsarService).getPulsarResources();
+ Awaitility.await().untilAsserted(() -> {
+ assertNotNull(resourceGroupConfigListener.getRgNamespaceConfigListener());
+ });
+ }
}
(pulsar) 01/02: [fix][broker] Fix ResourceGroup report local usage (#22340)
Posted by zi...@apache.org.
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
commit 36ba4766f614f5822326bfea718b68ee95029402
Author: Zixuan Liu <no...@gmail.com>
AuthorDate: Tue Mar 26 14:37:10 2024 +0800
[fix][broker] Fix ResourceGroup report local usage (#22340)
Signed-off-by: Zixuan Liu <no...@gmail.com>
(cherry picked from commit 0b2b6d593bb6ee9094d4c0a7a311490e7362f68f)
---
.../pulsar/broker/resourcegroup/ResourceGroup.java | 38 +++++---
.../broker/resourcegroup/ResourceGroupService.java | 4 +-
.../ResourceGroupReportLocalUsageTest.java | 105 +++++++++++++++++++++
3 files changed, 130 insertions(+), 17 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 0aa355eee7d..b4aacf0b1f0 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
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.broker.resourcegroup;
+import com.google.common.annotations.VisibleForTesting;
import io.prometheus.client.Counter;
import java.util.HashMap;
import java.util.Set;
@@ -218,24 +219,28 @@ public class ResourceGroup {
resourceUsage.setOwner(this.getID());
p = resourceUsage.setPublish();
- this.setUsageInMonitoredEntity(ResourceGroupMonitoringClass.Publish, p);
+ if (!this.setUsageInMonitoredEntity(ResourceGroupMonitoringClass.Publish, p)) {
+ resourceUsage.clearPublish();
+ }
p = resourceUsage.setDispatch();
- this.setUsageInMonitoredEntity(ResourceGroupMonitoringClass.Dispatch, p);
+ if (!this.setUsageInMonitoredEntity(ResourceGroupMonitoringClass.Dispatch, p)) {
+ resourceUsage.clearDispatch();
+ }
// Punt storage for now.
}
// Transport manager mandated op.
public void rgResourceUsageListener(String broker, ResourceUsage resourceUsage) {
- NetworkUsage p;
-
- p = resourceUsage.getPublish();
- this.getUsageFromMonitoredEntity(ResourceGroupMonitoringClass.Publish, p, broker);
-
- p = resourceUsage.getDispatch();
- this.getUsageFromMonitoredEntity(ResourceGroupMonitoringClass.Dispatch, p, broker);
+ if (resourceUsage.hasPublish()) {
+ this.getUsageFromMonitoredEntity(ResourceGroupMonitoringClass.Publish, resourceUsage.getPublish(), broker);
+ }
+ if (resourceUsage.hasDispatch()) {
+ this.getUsageFromMonitoredEntity(ResourceGroupMonitoringClass.Dispatch, resourceUsage.getDispatch(),
+ broker);
+ }
// Punt storage for now.
}
@@ -451,12 +456,6 @@ public class ResourceGroup {
bytesUsed = monEntity.usedLocallySinceLastReport.bytes;
messagesUsed = monEntity.usedLocallySinceLastReport.messages;
- monEntity.usedLocallySinceLastReport.bytes = monEntity.usedLocallySinceLastReport.messages = 0;
-
- monEntity.totalUsedLocally.bytes += bytesUsed;
- monEntity.totalUsedLocally.messages += messagesUsed;
-
- monEntity.lastResourceUsageFillTimeMSecsSinceEpoch = System.currentTimeMillis();
if (sendReport) {
p.setBytesPerPeriod(bytesUsed);
@@ -464,6 +463,10 @@ public class ResourceGroup {
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();
} else {
numSuppressions = monEntity.numSuppressedUsageReports++;
}
@@ -596,6 +599,11 @@ public class ResourceGroup {
};
}
+ @VisibleForTesting
+ PerMonitoringClassFields getMonitoredEntity(ResourceGroupMonitoringClass monClass) {
+ return this.monitoringClassFields[monClass.ordinal()];
+ }
+
public final String resourceGroupName;
public PerMonitoringClassFields[] monitoringClassFields =
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupService.java
index 48419d7c451..04cff21434e 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupService.java
@@ -686,7 +686,7 @@ public class ResourceGroupService implements AutoCloseable{
timeUnitScale);
this.resourceUsagePublishPeriodInSeconds = newPeriodInSeconds;
maxIntervalForSuppressingReportsMSecs =
- this.resourceUsagePublishPeriodInSeconds * MaxUsageReportSuppressRounds;
+ TimeUnit.SECONDS.toMillis(this.resourceUsagePublishPeriodInSeconds) * MaxUsageReportSuppressRounds;
}
}
@@ -705,7 +705,7 @@ public class ResourceGroupService implements AutoCloseable{
periodInSecs,
this.timeUnitScale);
maxIntervalForSuppressingReportsMSecs =
- this.resourceUsagePublishPeriodInSeconds * MaxUsageReportSuppressRounds;
+ TimeUnit.SECONDS.toMillis(this.resourceUsagePublishPeriodInSeconds) * MaxUsageReportSuppressRounds;
}
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
new file mode 100644
index 00000000000..658b7c94165
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupReportLocalUsageTest.java
@@ -0,0 +1,105 @@
+/*
+ * 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.resourcegroup;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
+import org.apache.pulsar.broker.resourcegroup.ResourceGroup.BytesAndMessagesCount;
+import org.apache.pulsar.broker.resourcegroup.ResourceGroup.PerMonitoringClassFields;
+import org.apache.pulsar.broker.resourcegroup.ResourceGroup.ResourceGroupMonitoringClass;
+import org.apache.pulsar.broker.service.resource.usage.ResourceUsage;
+import org.apache.pulsar.common.policies.data.ResourceGroup;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+public class ResourceGroupReportLocalUsageTest extends MockedPulsarServiceBaseTest {
+
+ @BeforeClass
+ @Override
+ protected void setup() throws Exception {
+ super.internalSetup();
+ }
+
+ @AfterClass(alwaysRun = true)
+ @Override
+ protected void cleanup() throws Exception {
+ super.internalCleanup();
+ }
+
+
+ @Test
+ public void testRgFillResourceUsage() throws Exception {
+ pulsar.getResourceGroupServiceManager().close();
+ AtomicBoolean needReport = new AtomicBoolean(false);
+ ResourceGroupService service = new ResourceGroupService(pulsar, TimeUnit.HOURS, null,
+ new ResourceQuotaCalculator() {
+ @Override
+ public boolean needToReportLocalUsage(long currentBytesUsed, long lastReportedBytes,
+ long currentMessagesUsed, long lastReportedMessages,
+ long lastReportTimeMSecsSinceEpoch) {
+ return needReport.get();
+ }
+
+ @Override
+ public long computeLocalQuota(long confUsage, long myUsage, long[] allUsages) {
+ return 0;
+ }
+ });
+ String rgName = "rg-1";
+ ResourceGroup rgConfig = new ResourceGroup();
+ rgConfig.setPublishRateInBytes(1000L);
+ 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);
+ ResourceUsage resourceUsage = new ResourceUsage();
+ resourceGroup.rgFillResourceUsage(resourceUsage);
+ assertFalse(resourceUsage.hasDispatch());
+ assertFalse(resourceUsage.hasPublish());
+
+ 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);
+
+ needReport.set(true);
+ 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);
+ }
+}
\ No newline at end of file