(rocketmq) branch develop updated: [ISSUE #9297] Supports outputting topic put TPS in TopicStatusSubCommand (#9298)

[lizhimin]

This is an automated email from the ASF dual-hosted git repository.

lizhimin pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git


The following commit(s) were added to refs/heads/develop by this push:
     new d5b474d9d3 [ISSUE #9297] Supports outputting topic put TPS in 
TopicStatusSubCommand (#9298)
d5b474d9d3 is described below

commit d5b474d9d326f380d736abe700f8a7a37fe364e2
Author: ymwneu <ymw...@126.com>
AuthorDate: Wed Apr 2 18:59:45 2025 +0800

    [ISSUE #9297] Supports outputting topic put TPS in TopicStatusSubCommand 
(#9298)
---
 .../apache/rocketmq/broker/processor/AdminBrokerProcessor.java |  1 +
 .../rocketmq/remoting/protocol/admin/TopicStatsTable.java      | 10 ++++++++++
 .../org/apache/rocketmq/store/stats/BrokerStatsManager.java    |  4 ++++
 .../org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java |  1 +
 .../rocketmq/tools/command/topic/TopicStatusSubCommand.java    |  2 ++
 5 files changed, 18 insertions(+)

diff --git 
a/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
 
b/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
index 4ff4bed814..4200f34bde 100644
--- 
a/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
+++ 
b/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
@@ -1837,6 +1837,7 @@ public class AdminBrokerProcessor implements 
NettyRequestProcessor {
                 topicStatsTable.getOffsetTable().put(mq, topicOffset);
             }
 
+            
topicStatsTable.setTopicPutTps(this.brokerController.getBrokerStatsManager().tpsTopicPutNums(requestHeader.getTopic()));
             byte[] body = topicStatsTable.encode();
             response.setBody(body);
             response.setCode(ResponseCode.SUCCESS);
diff --git 
a/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/admin/TopicStatsTable.java
 
b/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/admin/TopicStatsTable.java
index 9f467e7449..5cb2af6373 100644
--- 
a/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/admin/TopicStatsTable.java
+++ 
b/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/admin/TopicStatsTable.java
@@ -22,6 +22,8 @@ import org.apache.rocketmq.common.message.MessageQueue;
 import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
 
 public class TopicStatsTable extends RemotingSerializable {
+    private double topicPutTps;
+
     private Map<MessageQueue, TopicOffset> offsetTable = new 
ConcurrentHashMap<>();
 
     public Map<MessageQueue, TopicOffset> getOffsetTable() {
@@ -31,4 +33,12 @@ public class TopicStatsTable extends RemotingSerializable {
     public void setOffsetTable(Map<MessageQueue, TopicOffset> offsetTable) {
         this.offsetTable = offsetTable;
     }
+
+    public double getTopicPutTps() {
+        return topicPutTps;
+    }
+
+    public void setTopicPutTps(double topicPutTps) {
+        this.topicPutTps = topicPutTps;
+    }
 }
diff --git 
a/store/src/main/java/org/apache/rocketmq/store/stats/BrokerStatsManager.java 
b/store/src/main/java/org/apache/rocketmq/store/stats/BrokerStatsManager.java
index 530339c23b..c272a30234 100644
--- 
a/store/src/main/java/org/apache/rocketmq/store/stats/BrokerStatsManager.java
+++ 
b/store/src/main/java/org/apache/rocketmq/store/stats/BrokerStatsManager.java
@@ -595,6 +595,10 @@ public class BrokerStatsManager {
         this.statsTable.get(Stats.SNDBCK_PUT_NUMS).addValue(statsKey, 1, 1);
     }
 
+    public double tpsTopicPutNums(final String topic) {
+        return 
this.statsTable.get(TOPIC_PUT_NUMS).getStatsDataInMinute(topic).getTps();
+    }
+
     public double tpsGroupGetNums(final String group, final String topic) {
         final String statsKey = buildStatsKey(topic, group);
         return 
this.statsTable.get(Stats.GROUP_GET_NUMS).getStatsDataInMinute(statsKey).getTps();
diff --git 
a/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java
 
b/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java
index 9afd37f784..e6405cb2d9 100644
--- 
a/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java
+++ 
b/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java
@@ -367,6 +367,7 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, 
MQAdminExtInner {
                                 if (addr != null) {
                                     TopicStatsTable tst = 
mqClientInstance.getMQClientAPIImpl().getTopicStatsInfo(addr, topic, 
timeoutMillis);
                                     
topicStatsTable.getOffsetTable().putAll(tst.getOffsetTable());
+                                    
topicStatsTable.setTopicPutTps(topicStatsTable.getTopicPutTps() + 
tst.getTopicPutTps());
                                 }
                             } catch (Exception e) {
                                 logger.error("getTopicStatsInfo error. 
topic={}", topic, e);
diff --git 
a/tools/src/main/java/org/apache/rocketmq/tools/command/topic/TopicStatusSubCommand.java
 
b/tools/src/main/java/org/apache/rocketmq/tools/command/topic/TopicStatusSubCommand.java
index 47ca761d1f..ff91399f1c 100644
--- 
a/tools/src/main/java/org/apache/rocketmq/tools/command/topic/TopicStatusSubCommand.java
+++ 
b/tools/src/main/java/org/apache/rocketmq/tools/command/topic/TopicStatusSubCommand.java
@@ -113,6 +113,8 @@ public class TopicStatusSubCommand implements SubCommand {
                     humanTimestamp
                 );
             }
+            System.out.printf("%n");
+            System.out.printf("Topic Put TPS: %s%n", 
topicStatsTable.getTopicPutTps());
         } catch (Exception e) {
             throw new SubCommandException(this.getClass().getSimpleName() + " 
command failed", e);
         } finally {