merge 3.3

Author: fantiq

dubbo-common/src/test/java/org/apache/dubbo/common/bytecode/WrapperTest.java

@@ -18,6 +18,8 @@
 
 import org.apache.dubbo.common.utils.ClassUtils;
 
+import java.util.Arrays;
+
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.slf4j.Logger;
@@ -179,9 +181,14 @@ void test_getDeclaredMethodNames_ContainExtendsParentMethods() throws Exception
 
     @Test
     void test_getMethodNames_ContainExtendsParentMethods() throws Exception {
-        assertArrayEquals(
-                new String[] {"hello", "world"}, Wrapper.getWrapper(Son.class).getMethodNames());
-        assertArrayEquals(new String[] {"hello", "world"}, ClassUtils.getMethodNames(Son.class));
+        String[] methodNamesFromWrapepr = Wrapper.getWrapper(Son.class).getMethodNames();
+        String[] methodNamesFromClassUtils = ClassUtils.getMethodNames(Son.class);
+
+        Arrays.sort(methodNamesFromWrapepr);
+        Arrays.sort(methodNamesFromClassUtils);
+
+        assertArrayEquals(new String[] {"hello", "world"}, methodNamesFromWrapepr);
+        assertArrayEquals(new String[] {"hello", "world"}, methodNamesFromClassUtils);
     }
 
     @Test

dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/DefaultFuture.java

@@ -38,6 +38,7 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.TimeUnit;
 
 import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_TIMEOUT;
@@ -324,7 +325,12 @@ public void run(Timeout timeout) {
 
             ExecutorService executor = future.getExecutor();
             if (executor != null && !executor.isShutdown()) {
-                executor.execute(() -> notifyTimeout(future));
+                try {
+                    executor.execute(() -> notifyTimeout(future));
+                } catch (RejectedExecutionException e) {
+                    notifyTimeout(future);
+                    throw e;
+                }
             } else {
                 notifyTimeout(future);
             }

dubbo-remoting/dubbo-remoting-api/src/test/java/org/apache/dubbo/remoting/exchange/support/DefaultFutureTest.java

@@ -27,7 +27,10 @@
 
 import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
+import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.junit.jupiter.api.Assertions;
@@ -224,6 +227,43 @@ void testClose1() {
         Assertions.assertFalse(executor.isTerminated());
     }
 
+    @Test
+    void testTimeoutWithRejectedExecution() throws Exception {
+        // Create a ThreadPoolExecutor with a queue capacity of 1
+        ThreadPoolExecutor customExecutor = new ThreadPoolExecutor(
+                1, // corePoolSize
+                1, // maxPoolSize
+                60L,
+                TimeUnit.SECONDS,
+                new ArrayBlockingQueue<>(1), // queue capacity is 1
+                new ThreadPoolExecutor.AbortPolicy() // default rejection policy: throws exception
+                );
+        // Submit two tasks to occupy the thread and the queue
+        customExecutor.submit(() -> {
+            try {
+                Thread.sleep(500); // occupy the thread for a while
+            } catch (InterruptedException ignored) {
+            }
+        });
+        customExecutor.submit(() -> {
+            try {
+                Thread.sleep(500); // occupy the queue
+            } catch (InterruptedException ignored) {
+            }
+        });
+        // Create a Dubbo Mock Channel and a request
+        Channel channel = new MockedChannel();
+        Request request = new Request(999);
+        // Use Dubbo's newFuture and pass in the custom thread pool
+        DefaultFuture future = DefaultFuture.newFuture(channel, request, 100, customExecutor);
+        // Mark the request as sent
+        DefaultFuture.sent(channel, request);
+        // Wait for the timeout task to trigger
+        Thread.sleep(300);
+        Assertions.assertNull(DefaultFuture.getFuture(999), "Future should be removed from FUTURES after timeout");
+        customExecutor.shutdown();
+    }
+
     @Test
     void testClose2() {
         Channel channel = new MockedChannel();

dubbo-remoting/dubbo-remoting-zookeeper-curator5/src/main/java/org/apache/dubbo/remoting/zookeeper/curator5/Curator5ZookeeperClient.java

@@ -99,7 +99,10 @@ public List<ACL> getAclForPath(String path) {
             boolean connected = client.blockUntilConnected(timeout, TimeUnit.MILLISECONDS);
             boolean check = UrlUtils.isCheck(url);
             if (check && !connected) {
-                IllegalStateException illegalStateException = new IllegalStateException("zookeeper not connected");
+                // close CuratorFramework to stop re-connection.
+                client.close();
+                IllegalStateException illegalStateException =
+                        new IllegalStateException("zookeeper not connected, the address is: " + url);
 
                 // 5-1 Failed to connect to configuration center.
                 logger.error(

dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/tps/DefaultTPSLimiter.java

@@ -39,27 +39,32 @@ public class DefaultTPSLimiter implements TPSLimiter {
 
     @Override
     public boolean isAllowable(URL url, Invocation invocation) {
+        boolean isMethodLevelTpsConfigured =
+                url.hasMethodParameter(RpcUtils.getMethodName(invocation), TPS_LIMIT_RATE_KEY);
+        String key = isMethodLevelTpsConfigured
+                ? url.getServiceKey() + "#" + RpcUtils.getMethodName(invocation)
+                : url.getServiceKey();
         int rate = url.getMethodParameter(RpcUtils.getMethodName(invocation), TPS_LIMIT_RATE_KEY, -1);
         long interval = url.getMethodParameter(
                 RpcUtils.getMethodName(invocation), TPS_LIMIT_INTERVAL_KEY, DEFAULT_TPS_LIMIT_INTERVAL);
-        String serviceKey = url.getServiceKey();
+
         if (rate > 0) {
-            StatItem statItem = stats.get(serviceKey);
+            StatItem statItem = stats.get(key);
             if (statItem == null) {
-                stats.putIfAbsent(serviceKey, new StatItem(serviceKey, rate, interval));
-                statItem = stats.get(serviceKey);
+                stats.putIfAbsent(key, new StatItem(key, rate, interval));
+                statItem = stats.get(key);
             } else {
                 // rate or interval has changed, rebuild
                 if (statItem.getRate() != rate || statItem.getInterval() != interval) {
-                    stats.put(serviceKey, new StatItem(serviceKey, rate, interval));
-                    statItem = stats.get(serviceKey);
+                    stats.put(key, new StatItem(key, rate, interval));
+                    statItem = stats.get(key);
                 }
             }
             return statItem.isAllowable();
         } else {
-            StatItem statItem = stats.get(serviceKey);
+            StatItem statItem = stats.get(key);
             if (statItem != null) {
-                stats.remove(serviceKey);
+                stats.remove(key);
             }
         }
 

dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/filter/tps/DefaultTPSLimiterTest.java

@@ -65,12 +65,52 @@ void testIsNotAllowable() {
     }
 
     @Test
-    void testTPSLimiterForMethodLevelConfig() {
+    void testMethodLevelTpsOverridesServiceLevel() {
         Invocation invocation = new MockInvocation();
         URL url = URL.valueOf("test://test");
         url = url.addParameter(INTERFACE_KEY, "org.apache.dubbo.rpc.file.TpsService");
         url = url.addParameter(TPS_LIMIT_RATE_KEY, TEST_LIMIT_RATE);
         int tpsConfigForMethodLevel = 3;
+        url = url.addParameter("tps", 1);
+        url = url.addParameter("echo.tps", tpsConfigForMethodLevel);
+        url = url.addParameter(TPS_LIMIT_INTERVAL_KEY, 1000);
+        for (int i = 1; i <= tpsConfigForMethodLevel + 1; i++) {
+            if (i == tpsConfigForMethodLevel + 1) {
+                Assertions.assertFalse(defaultTPSLimiter.isAllowable(url, invocation));
+            } else {
+                Assertions.assertTrue(defaultTPSLimiter.isAllowable(url, invocation));
+            }
+        }
+    }
+
+    @Test
+    void testServiceLevelTpsWhenOtherMethodsHaveTps() {
+        Invocation invocation = new MockInvocation();
+        URL url = URL.valueOf("test://test");
+        url = url.addParameter(INTERFACE_KEY, "org.apache.dubbo.rpc.file.TpsService");
+        url = url.addParameter(TPS_LIMIT_RATE_KEY, TEST_LIMIT_RATE);
+        int tpsConfigForServiceLevel = 3;
+        url = url.addParameter("tps", tpsConfigForServiceLevel);
+        url = url.addParameter("otherMethod.tps", 1);
+        url = url.addParameter(TPS_LIMIT_INTERVAL_KEY, 1000);
+        for (int i = 1; i <= tpsConfigForServiceLevel + 1; i++) {
+            if (i == tpsConfigForServiceLevel + 1) {
+                Assertions.assertFalse(defaultTPSLimiter.isAllowable(url, invocation));
+            } else {
+                Assertions.assertTrue(defaultTPSLimiter.isAllowable(url, invocation));
+            }
+        }
+    }
+
+    @Test
+    void testMethodLevelTpsIsolation() {
+        Invocation invocation = new MockInvocation();
+        URL url = URL.valueOf("test://test");
+        url = url.addParameter(INTERFACE_KEY, "org.apache.dubbo.rpc.file.TpsService");
+        url = url.addParameter(TPS_LIMIT_RATE_KEY, TEST_LIMIT_RATE);
+        int tpsConfigForMethodLevel = 3;
+        url = url.addParameter("tps", 1);
+        url = url.addParameter("otherMethod.tps", 2);
         url = url.addParameter("echo.tps", tpsConfigForMethodLevel);
         url = url.addParameter(TPS_LIMIT_INTERVAL_KEY, 1000);
         for (int i = 1; i <= tpsConfigForMethodLevel + 1; i++) {

dubbo-test/dubbo-test-check/src/main/java/org/apache/dubbo/test/check/registrycenter/context/ZookeeperWindowsContext.java

@@ -49,6 +49,11 @@ public class ZookeeperWindowsContext extends ZookeeperContext {
      */
     private final ExecuteWatchdog WATCHDOG = new ExecuteWatchdog(ExecuteWatchdog.INFINITE_TIMEOUT);
 
+    /**
+     * Set it to TRUE when using WatchDog.
+     */
+    private boolean usedWatchDog = false;
+
     /**
      * The map to store the pair of clientPort and PID.
      */
@@ -93,6 +98,7 @@ public ExecutorService getExecutorService() {
      * Returns the {@link ExecuteWatchdog}.
      */
     public ExecuteWatchdog getWatchdog() {
+        usedWatchDog = true;
         return WATCHDOG;
     }
 
@@ -101,7 +107,10 @@ public ExecuteWatchdog getWatchdog() {
      */
     public void destroy() {
         this.processIds.clear();
-        this.WATCHDOG.destroyProcess();
+        // check WatchDog used flag to avoid hanging at destroyProcess when WatchDog is not used.
+        if (usedWatchDog) {
+            this.WATCHDOG.destroyProcess();
+        }
         try {
             DEFAULT_EXECUTOR_SERVICE.shutdownNow();
         } catch (SecurityException | NullPointerException ex) {

dubbo-test/dubbo-test-check/src/main/java/org/apache/dubbo/test/check/registrycenter/processor/ResetZookeeperProcessor.java

@@ -45,6 +45,8 @@ public void process(Context context) throws DubboTestException {
                 client.start();
                 boolean connected = client.blockUntilConnected(1000, TimeUnit.MILLISECONDS);
                 if (!connected) {
+                    // close CuratorFramework to stop re-connection.
+                    client.close();
                     throw new IllegalStateException("zookeeper not connected");
                 }
                 client.delete().deletingChildrenIfNeeded().forPath("/dubbo");