Reduce unnecessary switches

apache · Oct 23, 2024 · 632a107 · 632a107
1 parent 0984cf6
commit 632a107
Show file tree

Hide file tree

Showing 7 changed files with 21 additions and 62 deletions.
diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java b/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java
@@ -56,6 +56,7 @@
 import org.apache.rocketmq.client.producer.SendStatus;
 import org.apache.rocketmq.client.rpchook.NamespaceRpcHook;
 import org.apache.rocketmq.common.BoundaryType;
+import org.apache.rocketmq.common.CheckRocksdbCqWriteResult;
 import org.apache.rocketmq.common.MQVersion;
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.Pair;
@@ -113,7 +114,6 @@
 import org.apache.rocketmq.remoting.protocol.body.BrokerReplicasInfo;
 import org.apache.rocketmq.remoting.protocol.body.BrokerStatsData;
 import org.apache.rocketmq.remoting.protocol.body.CheckClientRequestBody;
-import org.apache.rocketmq.remoting.protocol.body.CheckRocksdbCqWriteProgressResponseBody;
 import org.apache.rocketmq.remoting.protocol.body.ClusterAclVersionInfo;
 import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
 import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult;
@@ -3019,7 +3019,7 @@ public QueryConsumeQueueResponseBody queryConsumeQueue(final String brokerAddr,
         throw new MQClientException(response.getCode(), response.getRemark());
     }
 
-    public CheckRocksdbCqWriteProgressResponseBody checkRocksdbCqWriteProgress(final String brokerAddr, final String topic, final long checkStoreTime, final boolean async, final long timeoutMillis) throws InterruptedException,
+    public CheckRocksdbCqWriteResult checkRocksdbCqWriteProgress(final String brokerAddr, final String topic, final long checkStoreTime, final boolean async, final long timeoutMillis) throws InterruptedException,
         RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, MQClientException {
         CheckRocksdbCqWriteProgressRequestHeader header = new CheckRocksdbCqWriteProgressRequestHeader();
         header.setTopic(topic);
@@ -3029,7 +3029,7 @@ public CheckRocksdbCqWriteProgressResponseBody checkRocksdbCqWriteProgress(final
         RemotingCommand response = this.remotingClient.invokeSync(brokerAddr, request, async ? timeoutMillis : timeoutMillis * 100);
         assert response != null;
         if (ResponseCode.SUCCESS == response.getCode()) {
-            return CheckRocksdbCqWriteProgressResponseBody.decode(response.getBody(), CheckRocksdbCqWriteProgressResponseBody.class);
+            return JSON.parseObject(response.getBody(), CheckRocksdbCqWriteResult.class);
         }
         throw new MQClientException(response.getCode(), response.getRemark());
     }

diff --git a/common/src/main/java/org/apache/rocketmq/common/CheckRocksdbCqWriteResult.java b/common/src/main/java/org/apache/rocketmq/common/CheckRocksdbCqWriteResult.java
@@ -26,7 +26,7 @@ public enum CheckStatus {
         CHECK_OK(0),
         CHECK_NOT_OK(1),
         CHECK_IN_PROGRESS(2),
-        CHECK_ERROR(4);
+        CHECK_ERROR(3);
 
         private int value;
 

diff --git a/...a/org/apache/rocketmq/remoting/protocol/body/CheckRocksdbCqWriteProgressResponseBody.java b/...a/org/apache/rocketmq/remoting/protocol/body/CheckRocksdbCqWriteProgressResponseBody.java
diff --git a/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExt.java b/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExt.java
@@ -26,6 +26,7 @@
 import org.apache.rocketmq.client.exception.MQBrokerException;
 import org.apache.rocketmq.client.exception.MQClientException;
 import org.apache.rocketmq.common.BoundaryType;
+import org.apache.rocketmq.common.CheckRocksdbCqWriteResult;
 import org.apache.rocketmq.common.Pair;
 import org.apache.rocketmq.common.PlainAccessConfig;
 import org.apache.rocketmq.common.TopicConfig;
@@ -46,7 +47,6 @@
 import org.apache.rocketmq.remoting.protocol.body.BrokerMemberGroup;
 import org.apache.rocketmq.remoting.protocol.body.BrokerReplicasInfo;
 import org.apache.rocketmq.remoting.protocol.body.BrokerStatsData;
-import org.apache.rocketmq.remoting.protocol.body.CheckRocksdbCqWriteProgressResponseBody;
 import org.apache.rocketmq.remoting.protocol.body.ClusterAclVersionInfo;
 import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
 import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult;
@@ -773,7 +773,7 @@ public QueryConsumeQueueResponseBody queryConsumeQueue(String brokerAddr, String
     }
 
     @Override
-    public CheckRocksdbCqWriteProgressResponseBody checkRocksdbCqWriteProgress(String brokerAddr, String topic, boolean async, long checkStoreTime)
+    public CheckRocksdbCqWriteResult checkRocksdbCqWriteProgress(String brokerAddr, String topic, boolean async, long checkStoreTime)
         throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, MQClientException {
         return this.defaultMQAdminExtImpl.checkRocksdbCqWriteProgress(brokerAddr, topic, async, checkStoreTime);
     }

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
@@ -46,6 +46,7 @@
 import org.apache.rocketmq.client.impl.MQClientManager;
 import org.apache.rocketmq.client.impl.factory.MQClientInstance;
 import org.apache.rocketmq.common.BoundaryType;
+import org.apache.rocketmq.common.CheckRocksdbCqWriteResult;
 import org.apache.rocketmq.common.KeyBuilder;
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.Pair;
@@ -84,7 +85,6 @@
 import org.apache.rocketmq.remoting.protocol.body.BrokerMemberGroup;
 import org.apache.rocketmq.remoting.protocol.body.BrokerReplicasInfo;
 import org.apache.rocketmq.remoting.protocol.body.BrokerStatsData;
-import org.apache.rocketmq.remoting.protocol.body.CheckRocksdbCqWriteProgressResponseBody;
 import org.apache.rocketmq.remoting.protocol.body.ClusterAclVersionInfo;
 import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
 import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult;
@@ -1819,7 +1819,7 @@ public QueryConsumeQueueResponseBody queryConsumeQueue(String brokerAddr, String
     }
 
     @Override
-    public CheckRocksdbCqWriteProgressResponseBody checkRocksdbCqWriteProgress(String brokerAddr, String topic, boolean async, long checkStoreTime)
+    public CheckRocksdbCqWriteResult checkRocksdbCqWriteProgress(String brokerAddr, String topic, boolean async, long checkStoreTime)
         throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, MQClientException {
         return this.mqClientInstance.getMQClientAPIImpl().checkRocksdbCqWriteProgress(brokerAddr, topic, checkStoreTime, async, timeoutMillis);
     }

diff --git a/tools/src/main/java/org/apache/rocketmq/tools/admin/MQAdminExt.java b/tools/src/main/java/org/apache/rocketmq/tools/admin/MQAdminExt.java
@@ -24,6 +24,7 @@
 import org.apache.rocketmq.client.MQAdmin;
 import org.apache.rocketmq.client.exception.MQBrokerException;
 import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.common.CheckRocksdbCqWriteResult;
 import org.apache.rocketmq.common.Pair;
 import org.apache.rocketmq.common.PlainAccessConfig;
 import org.apache.rocketmq.common.TopicConfig;
@@ -48,7 +49,6 @@
 import org.apache.rocketmq.remoting.protocol.body.ConsumeStatsList;
 import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
 import org.apache.rocketmq.remoting.protocol.body.ConsumerRunningInfo;
-import org.apache.rocketmq.remoting.protocol.body.CheckRocksdbCqWriteProgressResponseBody;
 import org.apache.rocketmq.remoting.protocol.body.EpochEntryCache;
 import org.apache.rocketmq.remoting.protocol.body.GroupList;
 import org.apache.rocketmq.remoting.protocol.body.HARuntimeInfo;
@@ -149,7 +149,7 @@ ConsumeStats examineConsumeStats(
         final String consumerGroup) throws RemotingException, MQClientException, InterruptedException,
         MQBrokerException;
 
-    CheckRocksdbCqWriteProgressResponseBody checkRocksdbCqWriteProgress(String brokerAddr, String topic, boolean async, long checkStoreTime)
+    CheckRocksdbCqWriteResult checkRocksdbCqWriteProgress(String brokerAddr, String topic, boolean async, long checkStoreTime)
         throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, MQClientException;
 
     ConsumeStats examineConsumeStats(final String consumerGroup,

diff --git a/...main/java/org/apache/rocketmq/tools/command/queue/CheckRocksdbCqWriteProgressCommand.java b/...main/java/org/apache/rocketmq/tools/command/queue/CheckRocksdbCqWriteProgressCommand.java
@@ -19,12 +19,13 @@
 
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 import org.apache.commons.cli.CommandLine;
 import org.apache.commons.cli.Option;
 import org.apache.commons.cli.Options;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.common.CheckRocksdbCqWriteResult;
 import org.apache.rocketmq.remoting.RPCHook;
-import org.apache.rocketmq.remoting.protocol.body.CheckRocksdbCqWriteProgressResponseBody;
 import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
 import org.apache.rocketmq.remoting.protocol.route.BrokerData;
 import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
@@ -72,8 +73,10 @@ public void execute(CommandLine commandLine, Options options, RPCHook rpcHook) {
         String clusterName = commandLine.hasOption('c') ? commandLine.getOptionValue('c').trim() : "";
         String topic = commandLine.hasOption('t') ? commandLine.getOptionValue('t').trim() : "";
         // The default check is 30 days
-        long checkStoreTime = commandLine.hasOption("checkFrom") ? Long.parseLong(commandLine.getOptionValue("checkFrom").trim()) : System.currentTimeMillis() - 24 * 60 * 60 * 1000 * 30L;
-        boolean async = commandLine.hasOption("asynchronous") ? Boolean.parseBoolean(commandLine.getOptionValue("asynchronous").trim()) : true;
+        long checkStoreTime = commandLine.hasOption("cf")
+            ? Long.parseLong(commandLine.getOptionValue("cf").trim())
+            : System.currentTimeMillis() - TimeUnit.DAYS.toMillis(30L);
+        boolean async = commandLine.hasOption("async") ? Boolean.parseBoolean(commandLine.getOptionValue("async").trim()) : true;
 
         try {
             defaultMQAdminExt.start();
@@ -88,12 +91,13 @@ public void execute(CommandLine commandLine, Options options, RPCHook rpcHook) {
                 String brokerName = entry.getKey();
                 BrokerData brokerData = entry.getValue();
                 String brokerAddr = brokerData.getBrokerAddrs().get(0L);
-                CheckRocksdbCqWriteProgressResponseBody body = defaultMQAdminExt.checkRocksdbCqWriteProgress(brokerAddr, topic, async, checkStoreTime);
+                CheckRocksdbCqWriteResult body = defaultMQAdminExt.checkRocksdbCqWriteProgress(brokerAddr, topic, async, checkStoreTime);
+                String tips = " checkStatus(0 -> ready, 1 -> notReady, 2-> checkDoing, 3 -> error): ";
                 if (StringUtils.isNotBlank(topic)) {
-                    System.out.print(brokerName + " checkStatus(0 -> ready, 1 -> notReady, 2 -> error): " + body.getCheckStatus() + " \n " + body.getDiffResult() + "\n");
+                    System.out.print(brokerName + tips + body.getCheckStatus() + " \n " + body.getCheckResult() + "\n");
                 } else {
-                    System.out.print(brokerName + " checkStatus(0 -> ready, 1 -> notReady, 2 -> error): " + body.getCheckStatus() + " \n " +
-                        body.getDiffResult() + "\n");
+                    System.out.print(brokerName + tips + body.getCheckStatus() + " \n " +
+                        body.getCheckResult() + "\n");
                 }
             }