This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch rocketmq-studio
in repository https://gitbox.apache.org/repos/asf/rocketmq-dashboard.git
The following commit(s) were added to refs/heads/rocketmq-studio by this push:
new c16d6c361 fix(cluster): read the legacy 4.x outbound TPS runtime stats
key (#3992)
c16d6c361 is described below
commit c16d6c3618fe7e42086a8ae715c1d37268f08608
Author: Zhao Jianing <[email protected]>
AuthorDate: Mon Sep 7 19:09:18 2026 +0800
fix(cluster): read the legacy 4.x outbound TPS runtime stats key (#3992)
RocketMQ brokers publish the outbound TPS under different runtime-stats
keys depending on the generation: 5.x emits "getTransferredTps" (double
r, see release-5.3.1 StoreStatsService) while 4.x keeps the historical
"getTransferedTps" spelling (see release-4.9.4 StoreStatsService).
The dashboard providers only read the 5.x key, so clusters backed by
4.x brokers report tpsOut = 0 on the Dashboard overview and the Cluster
page even when the broker is actively transferring messages. Probe both
spellings when resolving the outbound TPS entry.
Affected read sites:
- RocketMQDashboardProvider: stats.tpsOut and per-cluster tpsOut
- RocketMQClusterProvider: broker tpsOut
Regression covered by new tests that stub fetchBrokerRuntimeStats with
only the legacy key.
Signed-off-by: zjncs <[email protected]>
---
.../studio/common/util/BrokerRuntimeStats.java | 54 +++++++++++++++++++++
.../provider/apache/RocketMQClusterProvider.java | 3 +-
.../provider/apache/RocketMQDashboardProvider.java | 5 +-
.../studio/common/util/BrokerRuntimeStatsTest.java | 56 ++++++++++++++++++++++
.../apache/RocketMQClusterProviderTest.java | 17 +++++++
.../apache/RocketMQDashboardProviderTest.java | 19 ++++++++
6 files changed, 151 insertions(+), 3 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/common/util/BrokerRuntimeStats.java
b/server/src/main/java/org/apache/rocketmq/studio/common/util/BrokerRuntimeStats.java
new file mode 100644
index 000000000..925440cbe
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/common/util/BrokerRuntimeStats.java
@@ -0,0 +1,54 @@
+/*
+ * 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.rocketmq.studio.common.util;
+
+import org.springframework.util.StringUtils;
+
+import java.util.Map;
+
+/**
+ * Helpers for reading broker runtime-stats ({@code KVTable}) entries whose
keys differ across
+ * broker generations. Consolidates the per-provider copies so the legacy-key
fallback stays
+ * consistent everywhere runtime stats are parsed.
+ */
+public final class BrokerRuntimeStats {
+
+ /** 5.x brokers publish outbound TPS under this key (double r). */
+ private static final String OUTBOUND_TPS_5X = "getTransferredTps";
+ /** 4.x brokers use the historical single-r spelling. */
+ private static final String OUTBOUND_TPS_4X = "getTransferedTps";
+
+ private BrokerRuntimeStats() {
+ }
+
+ /**
+ * Resolves the outbound TPS entry: 5.x brokers emit {@code
getTransferredTps} while 4.x brokers
+ * emit the historical {@code getTransferedTps} spelling. A
present-but-blank 5.x value falls
+ * back to the 4.x key rather than being treated as the answer. Returns
null when neither key
+ * carries a usable value.
+ */
+ public static String outboundTps(Map<String, String> runtimeStats) {
+ if (runtimeStats == null) {
+ return null;
+ }
+ String transferred = runtimeStats.get(OUTBOUND_TPS_5X);
+ if (StringUtils.hasText(transferred)) {
+ return transferred;
+ }
+ return runtimeStats.get(OUTBOUND_TPS_4X);
+ }
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQClusterProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQClusterProvider.java
index e6d12fbdf..1e6332a02 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQClusterProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQClusterProvider.java
@@ -32,6 +32,7 @@ import
org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.common.domain.enums.BrokerStatus;
import org.apache.rocketmq.studio.common.domain.enums.ClusterStatus;
import org.apache.rocketmq.studio.common.domain.enums.ClusterType;
+import org.apache.rocketmq.studio.common.util.BrokerRuntimeStats;
import org.apache.rocketmq.tools.admin.MQAdminExt;
import org.springframework.context.annotation.Primary;
import org.springframework.stereotype.Service;
@@ -266,7 +267,7 @@ public class RocketMQClusterProvider implements
ClusterProvider {
builder.tpsIn(parseTpsValue(putTps));
}
- String getTransferredTps = table.get("getTransferredTps");
+ String getTransferredTps = BrokerRuntimeStats.outboundTps(table);
if (getTransferredTps != null && !getTransferredTps.isEmpty()) {
builder.tpsOut(parseTpsValue(getTransferredTps));
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQDashboardProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQDashboardProvider.java
index 7299fb0d0..00979692c 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQDashboardProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQDashboardProvider.java
@@ -44,6 +44,7 @@ import
org.apache.rocketmq.studio.ops.dashboard.ClusterOverviewVO;
import org.apache.rocketmq.studio.ops.dashboard.DashboardDataVO;
import org.apache.rocketmq.studio.ops.dashboard.DashboardProvider;
import org.apache.rocketmq.studio.ops.dashboard.DashboardStatsVO;
+import org.apache.rocketmq.studio.common.util.BrokerRuntimeStats;
import org.apache.rocketmq.studio.common.util.SystemGroupFilter;
import org.apache.rocketmq.studio.common.util.SystemTopicFilter;
import org.apache.rocketmq.tools.admin.MQAdminExt;
@@ -342,7 +343,7 @@ public class RocketMQDashboardProvider implements
DashboardProvider {
if (runtimeInfo != null && runtimeInfo.getTable() != null)
{
Map<String, String> table = runtimeInfo.getTable();
tpsIn += parseTps(table.get("putTps"));
- tpsOut += parseTps(table.get("getTransferredTps"));
+ tpsOut +=
parseTps(BrokerRuntimeStats.outboundTps(table));
messagesToday += parseMessagesToday(table);
}
@@ -374,7 +375,7 @@ public class RocketMQDashboardProvider implements
DashboardProvider {
KVTable runtimeInfo =
runtimeStatsByBroker.get(masterAddr);
if (runtimeInfo != null && runtimeInfo.getTable()
!= null) {
clusterTpsIn +=
parseTps(runtimeInfo.getTable().get("putTps"));
- clusterTpsOut +=
parseTps(runtimeInfo.getTable().get("getTransferredTps"));
+ clusterTpsOut +=
parseTps(BrokerRuntimeStats.outboundTps(runtimeInfo.getTable()));
String value =
runtimeInfo.getTable().get("brokerVersionDesc");
if (value != null &&
"unknown".equals(version)) {
String brokerVersion = value.trim();
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/common/util/BrokerRuntimeStatsTest.java
b/server/src/test/java/org/apache/rocketmq/studio/common/util/BrokerRuntimeStatsTest.java
new file mode 100644
index 000000000..0ea1c4c78
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/common/util/BrokerRuntimeStatsTest.java
@@ -0,0 +1,56 @@
+/*
+ * 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.rocketmq.studio.common.util;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class BrokerRuntimeStatsTest {
+
+ @Test
+ void prefersFiveXKeyWhenPresentTest() {
+ Map<String, String> stats = new HashMap<>();
+ stats.put("getTransferredTps", "5.0 6.0 7.0");
+ stats.put("getTransferedTps", "4.0 5.0 6.0");
+ assertThat(BrokerRuntimeStats.outboundTps(stats)).isEqualTo("5.0 6.0
7.0");
+ }
+
+ @Test
+ void fallsBackToFourXKeyWhenFiveXAbsentTest() {
+ Map<String, String> stats = new HashMap<>();
+ stats.put("getTransferedTps", "4.0 5.0 6.0");
+ assertThat(BrokerRuntimeStats.outboundTps(stats)).isEqualTo("4.0 5.0
6.0");
+ }
+
+ @Test
+ void fallsBackToFourXKeyWhenFiveXBlankTest() {
+ Map<String, String> stats = new HashMap<>();
+ stats.put("getTransferredTps", " ");
+ stats.put("getTransferedTps", "4.0 5.0 6.0");
+ assertThat(BrokerRuntimeStats.outboundTps(stats)).isEqualTo("4.0 5.0
6.0");
+ }
+
+ @Test
+ void returnsNullWhenNeitherKeyUsableTest() {
+ assertThat(BrokerRuntimeStats.outboundTps(new HashMap<>())).isNull();
+ assertThat(BrokerRuntimeStats.outboundTps(null)).isNull();
+ }
+}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQClusterProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQClusterProviderTest.java
index bdfe2a6f2..679bc5a0d 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQClusterProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQClusterProviderTest.java
@@ -73,6 +73,23 @@ class RocketMQClusterProviderTest {
assertThat(clusters.get(0).getStatus()).isEqualTo(ClusterStatus.healthy);
}
+ @Test
+ void discoverClustersShouldReadLegacy4xBrokerOutboundTpsKey() throws
Exception {
+ DefaultMQAdminExt adminExt = mock(DefaultMQAdminExt.class);
+ RocketMQClusterProvider provider = newProvider(adminExt);
+
+ when(adminExt.examineBrokerClusterInfo()).thenReturn(clusterInfo());
+ KVTable runtime = runtimeStats(" 12.7 10.0 9.0", null);
+ runtime.getTable().put("getTransferedTps", "34.2 30.0 29.0");
+
when(adminExt.fetchBrokerRuntimeStats("10.0.0.11:10911")).thenReturn(runtime);
+
+ List<ClusterVO> clusters = provider.discoverClusters();
+
+ assertThat(clusters).hasSize(1);
+
assertThat(clusters.get(0).getBrokers().get(0).getTpsIn()).isEqualTo(10);
+
assertThat(clusters.get(0).getBrokers().get(0).getTpsOut()).isEqualTo(30);
+ }
+
@Test
void discoverClustersShouldConvertDiskRatioToPercentage() throws Exception
{
DefaultMQAdminExt adminExt = mock(DefaultMQAdminExt.class);
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQDashboardProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQDashboardProviderTest.java
index 04c625f01..71dc46f8e 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQDashboardProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQDashboardProviderTest.java
@@ -484,6 +484,25 @@ class RocketMQDashboardProviderTest {
});
}
+ @Test
+ void dashboardShouldReadLegacy4xBrokerOutboundTpsKey() throws Exception {
+ DefaultMQAdminExt adminExt = mock(DefaultMQAdminExt.class);
+ when(adminExt.examineBrokerClusterInfo()).thenReturn(clusterInfo());
+ when(adminExt.fetchAllTopicList()).thenReturn(topicList());
+ KVTable runtime = runtimeStats("2.0", "5.0");
+ runtime.getTable().remove("getTransferredTps");
+ runtime.getTable().put("getTransferedTps", "4.0 5.0 6.0");
+
when(adminExt.fetchBrokerRuntimeStats("10.0.0.11:10911")).thenReturn(runtime);
+
+ DashboardDataVO dashboard = newProvider(adminExt).getDashboardData();
+
+ assertThat(dashboard.getStats().getTpsIn()).isEqualTo(2);
+ assertThat(dashboard.getStats().getTpsOut()).isEqualTo(5);
+ assertThat(dashboard.getClusters()).singleElement()
+ .extracting(ClusterOverviewVO::getTpsOut)
+ .isEqualTo(5L);
+ }
+
@Test
void dashboardShouldIgnoreNonFiniteNegativeAndOverflowingTps() throws
Exception {
DefaultMQAdminExt adminExt = mock(DefaultMQAdminExt.class);
