[ISSUE #9297] Supports outputting topic put TPS in TopicStatusSubCommand (#9298)

Author: ymwneu

broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java

@@ -1837,6 +1837,7 @@ private RemotingCommand getTopicStatsInfo(ChannelHandlerContext ctx,
                 topicStatsTable.getOffsetTable().put(mq, topicOffset);
             }
 
+            topicStatsTable.setTopicPutTps(this.brokerController.getBrokerStatsManager().tpsTopicPutNums(requestHeader.getTopic()));
             byte[] body = topicStatsTable.encode();
             response.setBody(body);
             response.setCode(ResponseCode.SUCCESS);

remoting/src/main/java/org/apache/rocketmq/remoting/protocol/admin/TopicStatsTable.java

@@ -22,6 +22,8 @@
 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 Map<MessageQueue, TopicOffset> getOffsetTable() {
     public void setOffsetTable(Map<MessageQueue, TopicOffset> offsetTable) {
         this.offsetTable = offsetTable;
     }
+
+    public double getTopicPutTps() {
+        return topicPutTps;
+    }
+
+    public void setTopicPutTps(double topicPutTps) {
+        this.topicPutTps = topicPutTps;
+    }
 }

store/src/main/java/org/apache/rocketmq/store/stats/BrokerStatsManager.java

@@ -595,6 +595,10 @@ public void incSendBackNums(final String group, final String topic) {
         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();

tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java

@@ -367,6 +367,7 @@ public void run() {
                                 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);

tools/src/main/java/org/apache/rocketmq/tools/command/topic/TopicStatusSubCommand.java

@@ -113,6 +113,8 @@ public void execute(final CommandLine commandLine, final Options options,
                     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 {