From 5a23fd07ddbeb7e78d2f8f87077b096da845b80d Mon Sep 17 00:00:00 2001
From: zjncs <18910855655@163.com>
Date: Tue, 8 Sep 2026 14:15:12 +0800
Subject: [PATCH] fix(remoting): tolerate missing properties in
 ConsumerRunningInfo analyze methods

The static analyze helpers of ConsumerRunningInfo consume data reported
by each consumer over the wire, so the well-known properties cannot be
assumed to exist:

- analyzeSubscription turns an absent PROP_CONSUMER_START_TIMESTAMP into
  the string "null" via String.valueOf and dies with
  NumberFormatException (For input string: "null"), aborting
  mqadmin consumerStatus midway.
- isPushType and the inlined copy in analyzeProcessQueue cast the raw
  Properties.get() value to ConsumeType and dereference it, throwing
  NullPointerException when PROP_CONSUME_TYPE is absent.

Keep the startup grace when the start timestamp is unknown (the 2-minute
window exists to tolerate freshly started consumers, and an unknown
start time deserves the same grace), return false from isPushType when
the consume type is unknown, and reuse isPushType in analyzeProcessQueue
to remove the duplicated block.

Signed-off-by: zjncs <18910855655@163.com>
---
 .../protocol/body/ConsumerRunningInfo.java    | 25 ++++-------
 .../body/ConsumerRunningInfoTest.java         | 42 +++++++++++++++++++
 2 files changed, 50 insertions(+), 17 deletions(-)

diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/body/ConsumerRunningInfo.java b/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/body/ConsumerRunningInfo.java
index 542f9300678..daa0fbe62eb 100644
--- a/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/body/ConsumerRunningInfo.java
+++ b/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/body/ConsumerRunningInfo.java
@@ -57,11 +57,10 @@ public static boolean analyzeSubscription(final TreeMap<String/* clientId */, Co
         boolean startForAWhile = false;
         {
 
-            String property = prev.getProperties().getProperty(ConsumerRunningInfo.PROP_CONSUMER_START_TIMESTAMP);
-            if (property == null) {
-                property = String.valueOf(prev.getProperties().get(ConsumerRunningInfo.PROP_CONSUMER_START_TIMESTAMP));
+            Object startTimestamp = prev.getProperties().get(ConsumerRunningInfo.PROP_CONSUMER_START_TIMESTAMP);
+            if (startTimestamp != null) {
+                startForAWhile = (System.currentTimeMillis() - Long.parseLong(String.valueOf(startTimestamp))) > (1000 * 60 * 2);
             }
-            startForAWhile = (System.currentTimeMillis() - Long.parseLong(property)) > (1000 * 60 * 2);
         }
 
         if (push && startForAWhile) {
@@ -96,12 +95,12 @@ public static boolean analyzeSubscription(final TreeMap<String/* clientId */, Co
     }
 
     public static boolean isPushType(ConsumerRunningInfo consumerRunningInfo) {
-        String property = consumerRunningInfo.getProperties().getProperty(ConsumerRunningInfo.PROP_CONSUME_TYPE);
+        Object consumeType = consumerRunningInfo.getProperties().get(ConsumerRunningInfo.PROP_CONSUME_TYPE);
 
-        if (property == null) {
-            property = ((ConsumeType) consumerRunningInfo.getProperties().get(ConsumerRunningInfo.PROP_CONSUME_TYPE)).name();
+        if (consumeType == null) {
+            return false;
         }
-        return ConsumeType.valueOf(property) == ConsumeType.CONSUME_PASSIVELY;
+        return ConsumeType.valueOf(consumeType.toString()) == ConsumeType.CONSUME_PASSIVELY;
     }
 
     public static boolean analyzeRebalance(final TreeMap<String/* clientId */, ConsumerRunningInfo> criTable) {
@@ -110,15 +109,7 @@ public static boolean analyzeRebalance(final TreeMap<String/* clientId */, Consu
 
     public static String analyzeProcessQueue(final String clientId, ConsumerRunningInfo info) {
         StringBuilder sb = new StringBuilder();
-        boolean push = false;
-        {
-            String property = info.getProperties().getProperty(ConsumerRunningInfo.PROP_CONSUME_TYPE);
-
-            if (property == null) {
-                property = ((ConsumeType) info.getProperties().get(ConsumerRunningInfo.PROP_CONSUME_TYPE)).name();
-            }
-            push = ConsumeType.valueOf(property) == ConsumeType.CONSUME_PASSIVELY;
-        }
+        boolean push = isPushType(info);
 
         boolean orderMsg = false;
         {
diff --git a/remoting/src/test/java/org/apache/rocketmq/remoting/protocol/body/ConsumerRunningInfoTest.java b/remoting/src/test/java/org/apache/rocketmq/remoting/protocol/body/ConsumerRunningInfoTest.java
index f05de389df3..b311d40b7a4 100644
--- a/remoting/src/test/java/org/apache/rocketmq/remoting/protocol/body/ConsumerRunningInfoTest.java
+++ b/remoting/src/test/java/org/apache/rocketmq/remoting/protocol/body/ConsumerRunningInfoTest.java
@@ -102,5 +102,47 @@ public void testAnalyzeSubscription() {
         assertThat(result).isTrue();
     }
 
+    @Test
+    public void testAnalyzeSubscriptionWithMissingStartTimestamp() {
+        // a client that does not report its start timestamp keeps the startup grace
+        // instead of failing the whole analysis
+        Properties properties = new Properties();
+        properties.put(ConsumerRunningInfo.PROP_CONSUME_TYPE, ConsumeType.CONSUME_PASSIVELY.name());
+        consumerRunningInfo.setProperties(properties);
+
+        ConsumerRunningInfo another = new ConsumerRunningInfo();
+        another.setProperties(new Properties(properties));
+        another.setSubscriptionSet(new TreeSet<>());
+
+        TreeMap<String, ConsumerRunningInfo> table = new TreeMap<>();
+        table.put("client_id_1", consumerRunningInfo);
+        table.put("client_id_2", another);
+
+        assertThat(ConsumerRunningInfo.analyzeSubscription(table)).isTrue();
+    }
+
+    @Test
+    public void testAnalyzeMethodsTolerateMissingProperties() {
+        ConsumerRunningInfo empty = new ConsumerRunningInfo();
+        empty.setProperties(new Properties());
+
+        assertThat(ConsumerRunningInfo.isPushType(empty)).isFalse();
+        assertThat(ConsumerRunningInfo.analyzeProcessQueue("client_id", empty)).isEmpty();
+
+        TreeMap<String, ConsumerRunningInfo> table = new TreeMap<>();
+        table.put("client_id", empty);
+        assertThat(ConsumerRunningInfo.analyzeSubscription(table)).isTrue();
+    }
+
+    @Test
+    public void testIsPushType() {
+        assertThat(ConsumerRunningInfo.isPushType(consumerRunningInfo)).isFalse();
+
+        Properties properties = new Properties();
+        properties.put(ConsumerRunningInfo.PROP_CONSUME_TYPE, ConsumeType.CONSUME_PASSIVELY.name());
+        consumerRunningInfo.setProperties(properties);
+        assertThat(ConsumerRunningInfo.isPushType(consumerRunningInfo)).isTrue();
+    }
+
 
 }