[rocketmq] branch develop updated: [ISSUE #2516]: Fix the value of sendThreadPoolQueueHeadWaitTimeMills is 0 most of the time (#3509)

[ji...@apache.org]

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

jinrongtong 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 87b0be0  [ISSUE #2516]: Fix the value of sendThreadPoolQueueHeadWaitTimeMills is 0 most of the time (#3509)
87b0be0 is described below

commit 87b0be00012dff86727766081d4243eeee280699
Author: 彭小漪 <64...@qq.com>
AuthorDate: Wed Nov 24 10:44:42 2021 +0800

    [ISSUE #2516]: Fix the value of sendThreadPoolQueueHeadWaitTimeMills is 0 most of the time (#3509)
---
 .../apache/rocketmq/broker/BrokerController.java   | 13 +++++---
 .../rocketmq/broker/BrokerControllerTest.java      | 35 ++++++++++++++++++++++
 2 files changed, 44 insertions(+), 4 deletions(-)

diff --git a/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java b/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
index 3e146b7..bce21c5 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
@@ -33,6 +33,8 @@ import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import java.util.Optional;
+import java.util.Objects;
 import org.apache.rocketmq.acl.AccessValidator;
 import org.apache.rocketmq.broker.client.ClientHousekeepingService;
 import org.apache.rocketmq.broker.client.ConsumerIdsChangeListener;
@@ -650,10 +652,13 @@ public class BrokerController {
 
     public long headSlowTimeMills(BlockingQueue<Runnable> q) {
         long slowTimeMills = 0;
-        final Runnable peek = q.peek();
-        if (peek != null) {
-            RequestTask rt = BrokerFastFailure.castRunnable(peek);
-            slowTimeMills = rt == null ? 0 : this.messageStore.now() - rt.getCreateTimestamp();
+        Optional<RequestTask> op = q.stream()
+                .map(BrokerFastFailure::castRunnable)
+                .filter(Objects::nonNull)
+                .findFirst();
+        if (op.isPresent()) {
+            RequestTask rt = op.get();
+            slowTimeMills = this.messageStore.now() - rt.getCreateTimestamp();
         }
 
         if (slowTimeMills < 0) {
diff --git a/broker/src/test/java/org/apache/rocketmq/broker/BrokerControllerTest.java b/broker/src/test/java/org/apache/rocketmq/broker/BrokerControllerTest.java
index dae1335..e8442a4 100644
--- a/broker/src/test/java/org/apache/rocketmq/broker/BrokerControllerTest.java
+++ b/broker/src/test/java/org/apache/rocketmq/broker/BrokerControllerTest.java
@@ -18,10 +18,16 @@
 package org.apache.rocketmq.broker;
 
 import java.io.File;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.rocketmq.broker.latency.FutureTaskExt;
 import org.apache.rocketmq.common.BrokerConfig;
 import org.apache.rocketmq.common.UtilAll;
 import org.apache.rocketmq.remoting.netty.NettyClientConfig;
 import org.apache.rocketmq.remoting.netty.NettyServerConfig;
+import org.apache.rocketmq.remoting.netty.RequestTask;
 import org.apache.rocketmq.store.config.MessageStoreConfig;
 import org.junit.After;
 import org.junit.Ignore;
@@ -47,4 +53,33 @@ public class BrokerControllerTest {
     public void destroy() {
         UtilAll.deleteFile(new File(new MessageStoreConfig().getStorePathRootDir()));
     }
+
+    @Test
+    public void testHeadSlowTimeMills() throws Exception {
+        BrokerController brokerController = new BrokerController(
+                new BrokerConfig(),
+                new NettyServerConfig(),
+                new NettyClientConfig(),
+                new MessageStoreConfig());
+        brokerController.initialize();
+        BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>();
+
+        //create task is not instance of FutureTaskExt;
+        Runnable runnable = new Runnable() {
+            @Override
+            public void run() {
+
+            }
+        };
+        queue.add(runnable);
+
+        RequestTask requestTask = new RequestTask(runnable, null, null);
+        // the requestTask is not the head of queue;
+        queue.add(new FutureTaskExt<>(requestTask, null));
+
+        long headSlowTimeMills = 100;
+        TimeUnit.MILLISECONDS.sleep(headSlowTimeMills);
+        assertThat(brokerController.headSlowTimeMills(queue)).isGreaterThanOrEqualTo(headSlowTimeMills);
+        //Attention: if we use the previous version method BrokerController#headSlowTimeMills, it will return 0;
+    }
 }