opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-08-30 18:10:44 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #8463] Some statistical items should also be deleted to prevent memory leakage when a topic or group is deleted (#8464)

Browse Source 
* Some important statistical items should also be deleted to prevent memory leakage when a topic or group is deleted

* Add UTs
This commit is contained in:
rongtong
2024-08-01 10:04:11 +08:00
committed by GitHub
parent 69334a708c
commit 3696be0632
3 changed files with 30 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/src/main/java/org/apache/rocketmq/common/stats/Stats.java
+3
View File
@@ -44,4 +44,7 @@ public class Stats {
public static final String GROUP_GET_FALL_SIZE = "GROUP_GET_FALL_SIZE";
public static final String GROUP_GET_FALL_TIME = "GROUP_GET_FALL_TIME";
public static final String GROUP_GET_LATENCY = "GROUP_GET_LATENCY";
public static final String TOPIC_PUT_LATENCY = "TOPIC_PUT_LATENCY";
public static final String GROUP_ACK_NUMS = "GROUP_ACK_NUMS";
public static final String GROUP_CK_NUMS = "GROUP_CK_NUMS";
}
store/src/main/java/org/apache/rocketmq/store/stats/BrokerStatsManager.java
+14 -10
View File
@@ -69,9 +69,9 @@ public class BrokerStatsManager {
@Deprecated public static final String COMMERCIAL_PERM_FAILURES = Stats.COMMERCIAL_PERM_FAILURES;
// Send message latency
public static final String TOPIC_PUT_LATENCY = "TOPIC_PUT_LATENCY";
public static final String GROUP_ACK_NUMS = "GROUP_ACK_NUMS";
public static final String GROUP_CK_NUMS = "GROUP_CK_NUMS";
@Deprecated public static final String TOPIC_PUT_LATENCY = "TOPIC_PUT_LATENCY";
@Deprecated public static final String GROUP_ACK_NUMS = "GROUP_ACK_NUMS";
@Deprecated public static final String GROUP_CK_NUMS = "GROUP_CK_NUMS";
public static final String DLQ_PUT_NUMS = "DLQ_PUT_NUMS";
public static final String BROKER_ACK_NUMS = "BROKER_ACK_NUMS";
public static final String BROKER_CK_NUMS = "BROKER_CK_NUMS";
@@ -179,10 +179,10 @@ public class BrokerStatsManager {
this.statsTable.put(Stats.TOPIC_PUT_SIZE, new StatsItemSet(Stats.TOPIC_PUT_SIZE, this.scheduledExecutorService, log));
this.statsTable.put(Stats.GROUP_GET_NUMS, new StatsItemSet(Stats.GROUP_GET_NUMS, this.scheduledExecutorService, log));
this.statsTable.put(Stats.GROUP_GET_SIZE, new StatsItemSet(Stats.GROUP_GET_SIZE, this.scheduledExecutorService, log));
this.statsTable.put(GROUP_ACK_NUMS, new StatsItemSet(GROUP_ACK_NUMS, this.scheduledExecutorService, log));
this.statsTable.put(GROUP_CK_NUMS, new StatsItemSet(GROUP_CK_NUMS, this.scheduledExecutorService, log));
this.statsTable.put(Stats.GROUP_ACK_NUMS, new StatsItemSet(Stats.GROUP_ACK_NUMS, this.scheduledExecutorService, log));
this.statsTable.put(Stats.GROUP_CK_NUMS, new StatsItemSet(Stats.GROUP_CK_NUMS, this.scheduledExecutorService, log));
this.statsTable.put(Stats.GROUP_GET_LATENCY, new StatsItemSet(Stats.GROUP_GET_LATENCY, this.scheduledExecutorService, log));
this.statsTable.put(TOPIC_PUT_LATENCY, new StatsItemSet(TOPIC_PUT_LATENCY, this.scheduledExecutorService, log));
this.statsTable.put(Stats.TOPIC_PUT_LATENCY, new StatsItemSet(Stats.TOPIC_PUT_LATENCY, this.scheduledExecutorService, log));
this.statsTable.put(Stats.SNDBCK_PUT_NUMS, new StatsItemSet(Stats.SNDBCK_PUT_NUMS, this.scheduledExecutorService, log));
this.statsTable.put(DLQ_PUT_NUMS, new StatsItemSet(DLQ_PUT_NUMS, this.scheduledExecutorService, log));
this.statsTable.put(Stats.BROKER_PUT_NUMS, new StatsItemSet(Stats.BROKER_PUT_NUMS, this.scheduledExecutorService, log));
@@ -338,10 +338,13 @@ public class BrokerStatsManager {
}
this.statsTable.get(Stats.GROUP_GET_NUMS).delValueByPrefixKey(topic, "@");
this.statsTable.get(Stats.GROUP_GET_SIZE).delValueByPrefixKey(topic, "@");
this.statsTable.get(Stats.GROUP_CK_NUMS).delValueByPrefixKey(topic, "@");
this.statsTable.get(Stats.GROUP_ACK_NUMS).delValueByPrefixKey(topic, "@");
this.statsTable.get(Stats.QUEUE_GET_NUMS).delValueByPrefixKey(topic, "@");
this.statsTable.get(Stats.QUEUE_GET_SIZE).delValueByPrefixKey(topic, "@");
this.statsTable.get(Stats.SNDBCK_PUT_NUMS).delValueByPrefixKey(topic, "@");
this.statsTable.get(Stats.GROUP_GET_LATENCY).delValueByInfixKey(topic, "@");
this.statsTable.get(Stats.TOPIC_PUT_LATENCY).delValueBySuffixKey(topic, "@");
this.momentStatsItemSetFallSize.delValueByInfixKey(topic, "@");
this.momentStatsItemSetFallTime.delValueByInfixKey(topic, "@");
}
@@ -349,6 +352,8 @@ public class BrokerStatsManager {
public void onGroupDeleted(final String group) {
this.statsTable.get(Stats.GROUP_GET_NUMS).delValueBySuffixKey(group, "@");
this.statsTable.get(Stats.GROUP_GET_SIZE).delValueBySuffixKey(group, "@");
this.statsTable.get(Stats.GROUP_CK_NUMS).delValueBySuffixKey(group, "@");
this.statsTable.get(Stats.GROUP_ACK_NUMS).delValueBySuffixKey(group, "@");
if (enableQueueStat) {
this.statsTable.get(Stats.QUEUE_GET_NUMS).delValueBySuffixKey(group, "@");
this.statsTable.get(Stats.QUEUE_GET_SIZE).delValueBySuffixKey(group, "@");
@@ -434,12 +439,12 @@ public class BrokerStatsManager {
public void incGroupCkNums(final String group, final String topic, final int incValue) {
final String statsKey = buildStatsKey(topic, group);
this.statsTable.get(GROUP_CK_NUMS).addValue(statsKey, incValue, 1);
this.statsTable.get(Stats.GROUP_CK_NUMS).addValue(statsKey, incValue, 1);
}
public void incGroupAckNums(final String group, final String topic, final int incValue) {
final String statsKey = buildStatsKey(topic, group);
this.statsTable.get(GROUP_ACK_NUMS).addValue(statsKey, incValue, 1);
this.statsTable.get(Stats.GROUP_ACK_NUMS).addValue(statsKey, incValue, 1);
}
public String buildStatsKey(String topic, String group) {
@@ -509,9 +514,8 @@ public class BrokerStatsManager {
statsKey = new StringBuilder(6);
}
statsKey.append(queueId).append("@").append(topic);
this.statsTable.get(TOPIC_PUT_LATENCY).addValue(statsKey.toString(), incValue, 1);
this.statsTable.get(Stats.TOPIC_PUT_LATENCY).addValue(statsKey.toString(), incValue, 1);
}
public void incBrokerPutNums() {
this.statsTable.get(Stats.BROKER_PUT_NUMS).getAndCreateStatsItem(this.clusterName).getValue().add(1);
}
store/src/test/java/org/apache/rocketmq/store/stats/BrokerStatsManagerTest.java
+13
View File
@@ -24,6 +24,8 @@ import org.junit.Before;
import org.junit.Test;
import static org.apache.rocketmq.common.stats.Stats.BROKER_PUT_NUMS;
import static org.apache.rocketmq.common.stats.Stats.GROUP_ACK_NUMS;
import static org.apache.rocketmq.common.stats.Stats.GROUP_CK_NUMS;
import static org.apache.rocketmq.common.stats.Stats.GROUP_GET_FALL_SIZE;
import static org.apache.rocketmq.common.stats.Stats.GROUP_GET_FALL_TIME;
import static org.apache.rocketmq.common.stats.Stats.GROUP_GET_LATENCY;
@@ -34,6 +36,7 @@ import static org.apache.rocketmq.common.stats.Stats.QUEUE_GET_SIZE;
import static org.apache.rocketmq.common.stats.Stats.QUEUE_PUT_NUMS;
import static org.apache.rocketmq.common.stats.Stats.QUEUE_PUT_SIZE;
import static org.apache.rocketmq.common.stats.Stats.SNDBCK_PUT_NUMS;
import static org.apache.rocketmq.common.stats.Stats.TOPIC_PUT_LATENCY;
import static org.apache.rocketmq.common.stats.Stats.TOPIC_PUT_NUMS;
import static org.apache.rocketmq.common.stats.Stats.TOPIC_PUT_SIZE;
import static org.assertj.core.api.Assertions.assertThat;
@@ -139,8 +142,11 @@ public class BrokerStatsManagerTest {
brokerStatsManager.incTopicPutSize(TOPIC, 100);
brokerStatsManager.incQueuePutNums(TOPIC, QUEUE_ID);
brokerStatsManager.incQueuePutSize(TOPIC, QUEUE_ID, 100);
brokerStatsManager.incTopicPutLatency(TOPIC, QUEUE_ID, 10);
brokerStatsManager.incGroupGetNums(GROUP_NAME, TOPIC, 1);
brokerStatsManager.incGroupGetSize(GROUP_NAME, TOPIC, 100);
brokerStatsManager.incGroupCkNums(GROUP_NAME, TOPIC, 1);
brokerStatsManager.incGroupAckNums(GROUP_NAME, TOPIC, 1);
brokerStatsManager.incQueueGetNums(GROUP_NAME, TOPIC, QUEUE_ID, 1);
brokerStatsManager.incQueueGetSize(GROUP_NAME, TOPIC, QUEUE_ID, 100);
brokerStatsManager.incSendBackNums(GROUP_NAME, TOPIC);
@@ -162,6 +168,9 @@ public class BrokerStatsManagerTest {
Assert.assertNull(brokerStatsManager.getStatsItem(GROUP_GET_LATENCY, "1@" + TOPIC + "@" + GROUP_NAME));
Assert.assertNull(brokerStatsManager.getStatsItem(GROUP_GET_FALL_SIZE, "1@" + TOPIC + "@" + GROUP_NAME));
Assert.assertNull(brokerStatsManager.getStatsItem(GROUP_GET_FALL_TIME, "1@" + TOPIC + "@" + GROUP_NAME));
Assert.assertNull(brokerStatsManager.getStatsItem(GROUP_CK_NUMS, TOPIC + "@" + GROUP_NAME));
Assert.assertNull(brokerStatsManager.getStatsItem(GROUP_ACK_NUMS, TOPIC + "@" + GROUP_NAME));
Assert.assertNull(brokerStatsManager.getStatsItem(TOPIC_PUT_LATENCY, QUEUE_ID + "@" + TOPIC));
}
@Test
@@ -174,6 +183,8 @@ public class BrokerStatsManagerTest {
brokerStatsManager.incGroupGetLatency(GROUP_NAME, TOPIC, 1, 1);
brokerStatsManager.recordDiskFallBehindTime(GROUP_NAME, TOPIC, 1, 11L);
brokerStatsManager.recordDiskFallBehindSize(GROUP_NAME, TOPIC, 1, 11L);
brokerStatsManager.incGroupCkNums(GROUP_NAME, TOPIC, 1);
brokerStatsManager.incGroupAckNums(GROUP_NAME, TOPIC, 1);
brokerStatsManager.onGroupDeleted(GROUP_NAME);
@@ -185,6 +196,8 @@ public class BrokerStatsManagerTest {
Assert.assertNull(brokerStatsManager.getStatsItem(GROUP_GET_LATENCY, "1@" + TOPIC + "@" + GROUP_NAME));
Assert.assertNull(brokerStatsManager.getStatsItem(GROUP_GET_FALL_SIZE, "1@" + TOPIC + "@" + GROUP_NAME));
Assert.assertNull(brokerStatsManager.getStatsItem(GROUP_GET_FALL_TIME, "1@" + TOPIC + "@" + GROUP_NAME));
Assert.assertNull(brokerStatsManager.getStatsItem(GROUP_CK_NUMS, TOPIC + "@" + GROUP_NAME));
Assert.assertNull(brokerStatsManager.getStatsItem(GROUP_ACK_NUMS, TOPIC + "@" + GROUP_NAME));
}
@Test