2.7.8 Dev

dubbo-all/pom.xml

@@ -981,6 +981,14 @@
                                     </resource>
                                 </transformer>
 
+                                <!-- @since 2.7.8 -->
+                                <transformer
+                                        implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
+                                    <resource>
+                                        META-INF/dubbo/internal/org.apache.dubbo.metadata.MetadataServiceExporter
+                                    </resource>
+                                </transformer>
+
                             </transformers>
                             <filters>
                                 <filter>

dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/Cluster.java

@@ -17,7 +17,9 @@
 package org.apache.dubbo.rpc.cluster;
 
 import org.apache.dubbo.common.extension.Adaptive;
+import org.apache.dubbo.common.extension.ExtensionLoader;
 import org.apache.dubbo.common.extension.SPI;
+import org.apache.dubbo.common.utils.StringUtils;
 import org.apache.dubbo.rpc.Invoker;
 import org.apache.dubbo.rpc.RpcException;
 import org.apache.dubbo.rpc.cluster.support.FailoverCluster;
@@ -29,8 +31,9 @@
  * <a href="http://en.wikipedia.org/wiki/Fault-tolerant_system">Fault-Tolerant</a>
  *
  */
-@SPI(FailoverCluster.NAME)
+@SPI(Cluster.DEFAULT)
 public interface Cluster {
+    String DEFAULT = FailoverCluster.NAME;
 
     /**
      * Merge the directory invokers to a virtual invoker.
@@ -43,4 +46,14 @@ public interface Cluster {
     @Adaptive
     <T> Invoker<T> join(Directory<T> directory) throws RpcException;
 
+    static Cluster getCluster(String name) {
+        return getCluster(name, true);
+    }
+
+    static Cluster getCluster(String name, boolean wrap) {
+        if (StringUtils.isEmpty(name)) {
+            name = Cluster.DEFAULT;
+        }
+        return ExtensionLoader.getExtensionLoader(Cluster.class).getExtension(name, wrap);
+    }
 }

dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/ClusterInvoker.java

@@ -0,0 +1,38 @@
+/*
+ * 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.dubbo.rpc.cluster;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.rpc.Invoker;
+
+/**
+ * This is the final Invoker type referenced by the RPC proxy on Consumer side.
+ * <p>
+ * A ClusterInvoker holds a group of normal invokers, stored in a Directory, mapping to one Registry.
+ * The ClusterInvoker implementation usually provides LB or HA policies, like FailoverClusterInvoker.
+ * <p>
+ * In multi-registry subscription scenario, the final ClusterInvoker will referr to several sub ClusterInvokers, with each
+ * sub ClusterInvoker representing one Registry. Take ZoneAwareClusterInvoker as an example, it is specially customized for
+ * multi-registry use cases: first, pick up one ClusterInvoker, then do LB inside the chose ClusterInvoker.
+ *
+ * @param <T>
+ */
+public interface ClusterInvoker<T> extends Invoker<T> {
+    URL getRegistryUrl();
+
+    Directory<T> getDirectory();
+}

dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/directory/AbstractDirectory.java

@@ -60,8 +60,7 @@ public AbstractDirectory(URL url, RouterChain<T> routerChain) {
         }
 
         this.url = url.removeParameter(REFER_KEY).removeParameter(MONITOR_KEY);
-        this.consumerUrl = url.addParameters(StringUtils.parseQueryString(url.getParameterAndDecoded(REFER_KEY)))
-                .removeParameter(MONITOR_KEY);
+        this.consumerUrl = this.url.addParameters(StringUtils.parseQueryString(url.getParameterAndDecoded(REFER_KEY)));
 
         setRouterChain(routerChain);
     }

dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/condition/config/ListenableRouter.java

@@ -92,7 +92,7 @@ public <T> List<Invoker<T>> route(List<Invoker<T>> invokers, URL url, Invocation
 
     @Override
     public int getPriority() {
-        return DEFAULT_PRIORITY;
+        return priority;
     }
 
     @Override

dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/mock/MockInvokersSelector.java

@@ -36,7 +36,7 @@
 public class MockInvokersSelector extends AbstractRouter {
 
     public static final String NAME = "MOCK_ROUTER";
-    private static final int MOCK_INVOKERS_DEFAULT_PRIORITY = Integer.MIN_VALUE;
+    private static final int MOCK_INVOKERS_DEFAULT_PRIORITY = -100;
 
     public MockInvokersSelector() {
         this.priority = MOCK_INVOKERS_DEFAULT_PRIORITY;

dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/AbstractClusterInvoker.java

@@ -30,6 +30,7 @@
 import org.apache.dubbo.rpc.RpcContext;
 import org.apache.dubbo.rpc.RpcException;
 import org.apache.dubbo.rpc.RpcInvocation;
+import org.apache.dubbo.rpc.cluster.ClusterInvoker;
 import org.apache.dubbo.rpc.cluster.Directory;
 import org.apache.dubbo.rpc.cluster.LoadBalance;
 import org.apache.dubbo.rpc.support.RpcUtils;
@@ -49,7 +50,7 @@
 /**
  * AbstractClusterInvoker
  */
-public abstract class AbstractClusterInvoker<T> implements Invoker<T> {
+public abstract class AbstractClusterInvoker<T> implements ClusterInvoker<T> {
 
     private static final Logger logger = LoggerFactory.getLogger(AbstractClusterInvoker.class);
 

dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/registry/ZoneAwareClusterInvoker.java

@@ -23,6 +23,7 @@
 import org.apache.dubbo.rpc.Invoker;
 import org.apache.dubbo.rpc.Result;
 import org.apache.dubbo.rpc.RpcException;
+import org.apache.dubbo.rpc.cluster.ClusterInvoker;
 import org.apache.dubbo.rpc.cluster.Directory;
 import org.apache.dubbo.rpc.cluster.LoadBalance;
 import org.apache.dubbo.rpc.cluster.support.AbstractClusterInvoker;
@@ -59,24 +60,23 @@ public ZoneAwareClusterInvoker(Directory<T> directory) {
     public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
         // First, pick the invoker (XXXClusterInvoker) that comes from the local registry, distinguish by a 'preferred' key.
         for (Invoker<T> invoker : invokers) {
-            // FIXME, the invoker is a cluster invoker representing one Registry, so it will automatically wrapped by MockClusterInvoker.
-            MockClusterInvoker<T> mockClusterInvoker = (MockClusterInvoker<T>) invoker;
-            if (mockClusterInvoker.isAvailable() && mockClusterInvoker.getRegistryUrl()
+            ClusterInvoker<T> clusterInvoker = (ClusterInvoker<T>) invoker;
+            if (clusterInvoker.isAvailable() && clusterInvoker.getRegistryUrl()
                     .getParameter(REGISTRY_KEY + "." + PREFERRED_KEY, false)) {
-                return mockClusterInvoker.invoke(invocation);
+                return clusterInvoker.invoke(invocation);
             }
         }
 
         // providers in the registry with the same zone
-        String zone = (String) invocation.getAttachment(REGISTRY_ZONE);
+        String zone = invocation.getAttachment(REGISTRY_ZONE);
         if (StringUtils.isNotEmpty(zone)) {
             for (Invoker<T> invoker : invokers) {
-                MockClusterInvoker<T> mockClusterInvoker = (MockClusterInvoker<T>) invoker;
-                if (mockClusterInvoker.isAvailable() && zone.equals(mockClusterInvoker.getRegistryUrl().getParameter(REGISTRY_KEY + "." + ZONE_KEY))) {
-                    return mockClusterInvoker.invoke(invocation);
+                ClusterInvoker<T> clusterInvoker = (ClusterInvoker<T>) invoker;
+                if (clusterInvoker.isAvailable() && zone.equals(clusterInvoker.getRegistryUrl().getParameter(REGISTRY_KEY + "." + ZONE_KEY))) {
+                    return clusterInvoker.invoke(invocation);
                 }
             }
-            String force = (String) invocation.getAttachment(REGISTRY_ZONE_FORCE);
+            String force = invocation.getAttachment(REGISTRY_ZONE_FORCE);
             if (StringUtils.isNotEmpty(force) && "true".equalsIgnoreCase(force)) {
                 throw new IllegalStateException("No registry instance in zone or no available providers in the registry, zone: "
                         + zone
@@ -93,12 +93,14 @@ public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, L
 
         // If none of the invokers has a preferred signal or is picked by the loadbalancer, pick the first one available.
         for (Invoker<T> invoker : invokers) {
-            MockClusterInvoker<T> mockClusterInvoker = (MockClusterInvoker<T>) invoker;
-            if (mockClusterInvoker.isAvailable()) {
-                return mockClusterInvoker.invoke(invocation);
+            ClusterInvoker<T> clusterInvoker = (ClusterInvoker<T>) invoker;
+            if (clusterInvoker.isAvailable()) {
+                return clusterInvoker.invoke(invocation);
             }
         }
-        throw new RpcException("No provider available in " + invokers);
+
+        //if none available,just pick one
+        return invokers.get(0).invoke(invocation);
     }
 
-}
+}

dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/wrapper/MockClusterInvoker.java

@@ -27,6 +27,7 @@
 import org.apache.dubbo.rpc.Result;
 import org.apache.dubbo.rpc.RpcException;
 import org.apache.dubbo.rpc.RpcInvocation;
+import org.apache.dubbo.rpc.cluster.ClusterInvoker;
 import org.apache.dubbo.rpc.cluster.Directory;
 import org.apache.dubbo.rpc.support.MockInvoker;
 
@@ -35,7 +36,7 @@
 import static org.apache.dubbo.rpc.Constants.MOCK_KEY;
 import static org.apache.dubbo.rpc.cluster.Constants.INVOCATION_NEED_MOCK;
 
-public class MockClusterInvoker<T> implements Invoker<T> {
+public class MockClusterInvoker<T> implements ClusterInvoker<T> {
 
     private static final Logger logger = LoggerFactory.getLogger(MockClusterInvoker.class);
 
@@ -57,6 +58,11 @@ public URL getRegistryUrl() {
         return directory.getUrl();
     }
 
+    @Override
+    public Directory<T> getDirectory() {
+        return directory;
+    }
+
     @Override
     public boolean isAvailable() {
         return directory.isAvailable();

dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/AbstractClusterInvokerTest.java

@@ -227,7 +227,7 @@ public void testSelect_multiInvokers() throws Exception {
     public void testCloseAvailablecheck() {
         LoadBalance lb = mock(LoadBalance.class);
         Map<String, String> queryMap = StringUtils.parseQueryString(url.getParameterAndDecoded(REFER_KEY));
-        URL tmpUrl = url.addParameters(queryMap).removeParameter(MONITOR_KEY);
+        URL tmpUrl = url.addParameters(queryMap).removeParameter(REFER_KEY).removeParameter(MONITOR_KEY);
         given(lb.select(invokers, tmpUrl, invocation)).willReturn(invoker1);
         initlistsize5();
 

dubbo-common/src/main/java/org/apache/dubbo/common/config/configcenter/AbstractDynamicConfiguration.java

@@ -236,7 +236,7 @@ protected ThreadPoolExecutor initWorkersThreadPool(String threadPoolPrefixName,
                                                        int threadPoolSize,
                                                        long keepAliveTime) {
         return new ThreadPoolExecutor(threadPoolSize, threadPoolSize, keepAliveTime,
-                TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), new NamedThreadFactory(threadPoolPrefixName));
+                TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), new NamedThreadFactory(threadPoolPrefixName, true));
     }
 
     protected static String getThreadPoolPrefixName(URL url) {

dubbo-common/src/main/java/org/apache/dubbo/common/config/configcenter/file/FileSystemDynamicConfiguration.java

@@ -24,6 +24,7 @@
 import org.apache.dubbo.common.config.configcenter.TreePathDynamicConfiguration;
 import org.apache.dubbo.common.function.ThrowableConsumer;
 import org.apache.dubbo.common.function.ThrowableFunction;
+import org.apache.dubbo.common.lang.ShutdownHookCallbacks;
 import org.apache.dubbo.common.utils.NamedThreadFactory;
 import org.apache.dubbo.common.utils.StringUtils;
 
@@ -146,6 +147,7 @@ public class FileSystemDynamicConfiguration extends TreePathDynamicConfiguration
         MODIFIERS = initWatchEventModifiers();
         DELAY = initDelay(MODIFIERS);
         WATCH_EVENTS_LOOP_THREAD_POOL = newWatchEventsLoopThreadPool();
+        registerDubboShutdownHook();
     }
 
     /**
@@ -230,6 +232,24 @@ private void doInListener(String configFilePath, BiConsumer<File, List<Configura
         });
     }
 
+    /**
+     * Register the Dubbo ShutdownHook
+     *
+     * @since 2.7.8
+     */
+    private static void registerDubboShutdownHook() {
+        ShutdownHookCallbacks.INSTANCE.addCallback(() -> {
+            watchService.ifPresent(w -> {
+                try {
+                    w.close();
+                } catch (IOException e) {
+                    throw new RuntimeException(e);
+                }
+            });
+            getWatchEventsLoopThreadPool().shutdown();
+        });
+    }
+
     private static boolean isProcessingWatchEvents() {
         return getWatchEventsLoopThreadPool().getActiveCount() > 0;
     }