[ROCKETMQ-32] Improve concision - Reuse local variable 'brokerAddrs' in RouteInfoManager.getSystemTopicList method

NOTICE

@@ -1,5 +1,5 @@
-RocketMQ
-Copyright 2016 Alibaba Group.
+Apache RocketMQ (incubating)
+Copyright 2016 The Apache Software Foundation
 
 This product includes software developed at
 The Apache Software Foundation (http://www.apache.org/).

bin/mqbroker

@@ -42,6 +42,4 @@ fi
 
 export ROCKETMQ_HOME
 
-rm -f $HOME/rmq_bk_gc.log.bac
-cp $HOME/rmq_bk_gc.log $HOME/rmq_bk_gc.log.bac
 sh ${ROCKETMQ_HOME}/bin/runbroker.sh org.apache.rocketmq.broker.BrokerStartup $@

broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java

@@ -463,7 +463,7 @@ public long headSlowTimeMills(BlockingQueue<Runnable> q) {
         final Runnable peek = q.peek();
         if (peek != null) {
             RequestTask rt = BrokerFastFailure.castRunnable(peek);
-            slowTimeMills = this.messageStore.now() - rt.getCreateTimestamp();
+            slowTimeMills = rt == null ? 0 : this.messageStore.now() - rt.getCreateTimestamp();
         }
 
         if (slowTimeMills < 0)

broker/src/main/java/org/apache/rocketmq/broker/latency/BrokerFastFailure.java

@@ -71,7 +71,7 @@ private void cleanExpiredRequest() {
                 } else {
                     break;
                 }
-            } catch (Throwable e) {
+            } catch (Throwable ignored) {
             }
         }
 
@@ -99,7 +99,7 @@ private void cleanExpiredRequest() {
                 } else {
                     break;
                 }
-            } catch (Throwable e) {
+            } catch (Throwable ignored) {
             }
         }
     }

broker/src/main/java/org/apache/rocketmq/broker/longpolling/PullRequestHoldService.java

@@ -130,7 +130,7 @@ public void notifyMessageArriving(final String topic, final int queueId, final l
                     if (newestOffset > request.getPullFromThisOffset()) {
                         if (this.messageFilter.isMessageMatched(request.getSubscriptionData(), tagsCode)) {
                             try {
-                                this.brokerController.getPullMessageProcessor().excuteRequestWhenWakeup(request.getClientChannel(),
+                                this.brokerController.getPullMessageProcessor().executeRequestWhenWakeup(request.getClientChannel(),
                                     request.getRequestCommand());
                             } catch (Throwable e) {
                                 log.error("execute request when wakeup failed.", e);
@@ -141,7 +141,7 @@ public void notifyMessageArriving(final String topic, final int queueId, final l
 
                     if (System.currentTimeMillis() >= (request.getSuspendTimestamp() + request.getTimeoutMillis())) {
                         try {
-                            this.brokerController.getPullMessageProcessor().excuteRequestWhenWakeup(request.getClientChannel(),
+                            this.brokerController.getPullMessageProcessor().executeRequestWhenWakeup(request.getClientChannel(),
                                 request.getRequestCommand());
                         } catch (Throwable e) {
                             log.error("execute request when wakeup failed.", e);

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

@@ -50,6 +50,7 @@
 import org.apache.rocketmq.remoting.common.RemotingUtil;
 import org.apache.rocketmq.remoting.exception.RemotingCommandException;
 import org.apache.rocketmq.remoting.netty.NettyRequestProcessor;
+import org.apache.rocketmq.remoting.netty.RequestTask;
 import org.apache.rocketmq.remoting.protocol.RemotingCommand;
 import org.apache.rocketmq.store.GetMessageResult;
 import org.apache.rocketmq.store.MessageExtBrokerInner;
@@ -481,7 +482,7 @@ private void generateOffsetMovedEvent(final OffsetMovedEvent event) {
         }
     }
 
-    public void excuteRequestWhenWakeup(final Channel channel, final RemotingCommand request) throws RemotingCommandException {
+    public void executeRequestWhenWakeup(final Channel channel, final RemotingCommand request) throws RemotingCommandException {
         Runnable run = new Runnable() {
             @Override
             public void run() {
@@ -513,8 +514,7 @@ public void operationComplete(ChannelFuture future) throws Exception {
                 }
             }
         };
-
-        this.brokerController.getPullMessageExecutor().submit(run);
+        this.brokerController.getPullMessageExecutor().submit(new RequestTask(run, channel, request));
     }
 
     public void registerConsumeMessageHook(List<ConsumeMessageHook> sendMessageHookList) {

broker/src/main/java/org/apache/rocketmq/broker/subscription/SubscriptionGroupManager.java

@@ -16,7 +16,6 @@
  */
 package org.apache.rocketmq.broker.subscription;
 
-import java.io.File;
 import java.util.Iterator;
 import java.util.Map.Entry;
 import java.util.concurrent.ConcurrentHashMap;
@@ -142,8 +141,8 @@ public String encode() {
 
     @Override
     public String configFilePath() {
-        //return BrokerPathConfigHelper.getSubscriptionGroupPath(this.brokerController.getMessageStoreConfig().getStorePathRootDir());
-        return BrokerPathConfigHelper.getSubscriptionGroupPath(System.getProperty("user.home") + File.separator + "store");
+        return BrokerPathConfigHelper.getSubscriptionGroupPath(this.brokerController.getMessageStoreConfig()
+                .getStorePathRootDir());
     }
 
     @Override

broker/src/main/java/org/apache/rocketmq/broker/topic/TopicConfigManager.java

@@ -16,7 +16,6 @@
  */
 package org.apache.rocketmq.broker.topic;
 
-import java.io.File;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.Map;
@@ -381,9 +380,8 @@ public String encode() {
 
     @Override
     public String configFilePath() {
-//        return BrokerPathConfigHelper.getTopicConfigPath(this.brokerController.getMessageStoreConfig()
-//                .getStorePathRootDir());
-        return BrokerPathConfigHelper.getTopicConfigPath(System.getProperty("user.home") + File.separator + "store");
+        return BrokerPathConfigHelper.getTopicConfigPath(this.brokerController.getMessageStoreConfig()
+                .getStorePathRootDir());
     }
 
     @Override

broker/src/test/java/org/apache/rocketmq/broker/api/BrokerFastFailureTest.java

@@ -0,0 +1,61 @@
+/*
+ * 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.broker.api;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import org.apache.rocketmq.broker.BrokerTestHarness;
+import org.apache.rocketmq.broker.latency.BrokerFastFailure;
+import org.apache.rocketmq.broker.latency.FutureTaskExt;
+import org.apache.rocketmq.remoting.netty.RequestTask;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class BrokerFastFailureTest extends BrokerTestHarness {
+
+    @Test
+    public void testHeadSlowTimeMills() throws InterruptedException {
+        BlockingQueue<Runnable> blockingQueue = new LinkedBlockingQueue<>();
+        blockingQueue.add(new FutureTaskExt<>(new RequestTask(null, null, null), null));
+        TimeUnit.MILLISECONDS.sleep(10);
+        Assert.assertTrue(this.brokerController.headSlowTimeMills(blockingQueue) > 0);
+
+        blockingQueue.clear();
+        blockingQueue.add(new Runnable() {
+            @Override public void run() {
+
+            }
+        });
+        Assert.assertTrue(this.brokerController.headSlowTimeMills(blockingQueue) == 0);
+    }
+
+    @Test
+    public void testCastRunnable() {
+        Runnable runnable = new Runnable() {
+            @Override public void run() {
+
+            }
+        };
+        Assert.assertNull(BrokerFastFailure.castRunnable(runnable));
+
+        RequestTask requestTask = new RequestTask(null, null, null);
+        runnable = new FutureTaskExt<>(requestTask, null);
+
+        Assert.assertEquals(requestTask, BrokerFastFailure.castRunnable(runnable));
+    }
+}