diff --git a/compiler/pom.xml b/compiler/pom.xml index d062be160e..fe3c2ba1e0 100644 --- a/compiler/pom.xml +++ b/compiler/pom.xml @@ -103,7 +103,7 @@ release - 2.11.1 + 2.17.0 diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/filter/DefaultFilterChainBuilder.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/filter/DefaultFilterChainBuilder.java index eb4d35efe0..15be3004c2 100644 --- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/filter/DefaultFilterChainBuilder.java +++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/filter/DefaultFilterChainBuilder.java @@ -18,13 +18,21 @@ package org.apache.dubbo.rpc.cluster.filter; import org.apache.dubbo.common.URL; import org.apache.dubbo.common.extension.Activate; +import org.apache.dubbo.common.extension.ExtensionDirector; +import org.apache.dubbo.common.extension.support.MultiInstanceActivateComparator; import org.apache.dubbo.common.utils.CollectionUtils; import org.apache.dubbo.rpc.Filter; import org.apache.dubbo.rpc.Invoker; import org.apache.dubbo.rpc.cluster.ClusterInvoker; +import org.apache.dubbo.rpc.model.ApplicationModel; +import org.apache.dubbo.rpc.model.ModuleModel; +import org.apache.dubbo.rpc.model.ScopeModel; import org.apache.dubbo.rpc.model.ScopeModelUtil; +import java.util.ArrayList; import java.util.List; +import java.util.Map; +import java.util.TreeMap; @Activate public class DefaultFilterChainBuilder implements FilterChainBuilder { @@ -36,14 +44,32 @@ public class DefaultFilterChainBuilder implements FilterChainBuilder { public Invoker buildInvokerChain(final Invoker originalInvoker, String key, String group) { Invoker last = originalInvoker; URL url = originalInvoker.getUrl(); - List filters = ScopeModelUtil.getExtensionLoader(Filter.class, url.getScopeModel()).getActivateExtension(url, key, group); + List moduleModels = getModuleModelsFromUrl(url); + List filters; + if (moduleModels != null && moduleModels.size() == 1) { + filters = ScopeModelUtil.getExtensionLoader(Filter.class, moduleModels.get(0)).getActivateExtension(url, key, group); + } else if (moduleModels != null && moduleModels.size() > 1) { + filters = new ArrayList<>(); + List directors = new ArrayList<>(); + for (ModuleModel moduleModel : moduleModels) { + List tempFilters = ScopeModelUtil.getExtensionLoader(Filter.class, moduleModel).getActivateExtension(url, key, group); + filters.addAll(tempFilters); + directors.add(moduleModel.getExtensionDirector()); + } + filters = sortingAndDeduplication(filters, directors); + + } else { + filters = ScopeModelUtil.getExtensionLoader(Filter.class, null).getActivateExtension(url, key, group); + } + if (!CollectionUtils.isEmpty(filters)) { for (int i = filters.size() - 1; i >= 0; i--) { final Filter filter = filters.get(i); final Invoker next = last; - last = new FilterChainNode<>(originalInvoker, next, filter); + last = new CopyOfFilterChainNode<>(originalInvoker, next, filter); } + return new CallbackRegistrationInvoker<>(last, filters); } return last; @@ -56,17 +82,64 @@ public class DefaultFilterChainBuilder implements FilterChainBuilder { public ClusterInvoker buildClusterInvokerChain(final ClusterInvoker originalInvoker, String key, String group) { ClusterInvoker last = originalInvoker; URL url = originalInvoker.getUrl(); - List filters = ScopeModelUtil.getExtensionLoader(ClusterFilter.class, url.getScopeModel()).getActivateExtension(url, key, group); + List moduleModels = getModuleModelsFromUrl(url); + List filters; + if (moduleModels != null && moduleModels.size() == 1) { + filters = ScopeModelUtil.getExtensionLoader(ClusterFilter.class, moduleModels.get(0)).getActivateExtension(url, key, group); + } else if (moduleModels != null && moduleModels.size() > 1) { + filters = new ArrayList<>(); + List directors = new ArrayList<>(); + for (ModuleModel moduleModel : moduleModels) { + List tempFilters = ScopeModelUtil.getExtensionLoader(ClusterFilter.class, moduleModel).getActivateExtension(url, key, group); + filters.addAll(tempFilters); + directors.add(moduleModel.getExtensionDirector()); + } + filters = sortingAndDeduplication(filters, directors); + + } else { + filters = ScopeModelUtil.getExtensionLoader(ClusterFilter.class, null).getActivateExtension(url, key, group); + } if (!CollectionUtils.isEmpty(filters)) { for (int i = filters.size() - 1; i >= 0; i--) { final ClusterFilter filter = filters.get(i); final Invoker next = last; - last = new ClusterFilterChainNode<>(originalInvoker, next, filter); + last = new CopyOfClusterFilterChainNode<>(originalInvoker, next, filter); } + return new ClusterCallbackRegistrationInvoker<>(originalInvoker, last, filters); } return last; } + private List sortingAndDeduplication(List filters, List directors) { + Map, T> filtersSet = new TreeMap<>(new MultiInstanceActivateComparator(directors)); + for (T filter : filters) { + filtersSet.putIfAbsent(filter.getClass(), filter); + } + return new ArrayList<>(filtersSet.values()); + } + + /** + * When the application-level service registration and discovery strategy is adopted, the URL will be of type InstanceAddressURL, + * and InstanceAddressURL belongs to the application layer and holds the ApplicationModel, + * but the filter is at the module layer and holds the ModuleModel, + * so it needs to be based on the url in the ScopeModel type to parse out all the moduleModels held by the url + * to obtain the filter configuration. + * + * @param url URL + * @return All ModuleModels in the url + */ + private List getModuleModelsFromUrl(URL url) { + List moduleModels = null; + ScopeModel scopeModel = url.getScopeModel(); + if (scopeModel instanceof ApplicationModel) { + moduleModels = ((ApplicationModel) scopeModel).getPubModuleModels(); + } else if (scopeModel instanceof ModuleModel) { + moduleModels = new ArrayList<>(); + moduleModels.add((ModuleModel) scopeModel); + } + return moduleModels; + } + } diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/filter/FilterChainBuilder.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/filter/FilterChainBuilder.java index 7edab0899a..a329959e3d 100644 --- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/filter/FilterChainBuilder.java +++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/filter/FilterChainBuilder.java @@ -16,8 +16,11 @@ */ package org.apache.dubbo.rpc.cluster.filter; +import org.apache.dubbo.common.Experimental; import org.apache.dubbo.common.URL; import org.apache.dubbo.common.extension.SPI; +import org.apache.dubbo.common.logger.Logger; +import org.apache.dubbo.common.logger.LoggerFactory; import org.apache.dubbo.rpc.BaseFilter; import org.apache.dubbo.rpc.Filter; import org.apache.dubbo.rpc.Invocation; @@ -28,6 +31,9 @@ import org.apache.dubbo.rpc.RpcException; import org.apache.dubbo.rpc.cluster.ClusterInvoker; import org.apache.dubbo.rpc.cluster.Directory; +import java.util.List; +import java.util.stream.Collectors; + import static org.apache.dubbo.common.extension.ExtensionScope.APPLICATION; @SPI(value = "default", scope = APPLICATION) @@ -144,12 +150,208 @@ public interface FilterChainBuilder { * @param */ class ClusterFilterChainNode, FILTER extends BaseFilter> - extends FilterChainNode implements ClusterInvoker { + extends FilterChainNode implements ClusterInvoker { public ClusterFilterChainNode(TYPE originalInvoker, Invoker nextNode, FILTER filter) { super(originalInvoker, nextNode, filter); } + @Override + public URL getRegistryUrl() { + return getOriginalInvoker().getRegistryUrl(); + } + + @Override + public Directory getDirectory() { + return getOriginalInvoker().getDirectory(); + } + + @Override + public boolean isDestroyed() { + return getOriginalInvoker().isDestroyed(); + } + } + + class CallbackRegistrationInvoker implements Invoker { + static final Logger LOGGER = LoggerFactory.getLogger(CallbackRegistrationInvoker.class); + final Invoker filterInvoker; + final List filters; + + public CallbackRegistrationInvoker(Invoker filterInvoker, List filters) { + this.filterInvoker = filterInvoker; + this.filters = filters; + } + + @Override + public Result invoke(Invocation invocation) throws RpcException { + Result asyncResult = filterInvoker.invoke(invocation); + asyncResult.whenCompleteWithContext((r, t) -> { + for (int i = filters.size() - 1; i >= 0; i--) { + FILTER filter = filters.get(i); + try { + if (filter instanceof ListenableFilter) { + ListenableFilter listenableFilter = ((ListenableFilter) filter); + Filter.Listener listener = listenableFilter.listener(invocation); + try { + if (listener != null) { + if (t == null) { + listener.onResponse(r, filterInvoker, invocation); + } else { + listener.onError(t, filterInvoker, invocation); + } + } + } finally { + listenableFilter.removeListener(invocation); + } + } else if (filter instanceof FILTER.Listener) { + FILTER.Listener listener = (FILTER.Listener) filter; + if (t == null) { + listener.onResponse(r, filterInvoker, invocation); + } else { + listener.onError(t, filterInvoker, invocation); + } + } + } catch (Throwable filterThrowable) { + LOGGER.error(String.format("Exception occurred while executing the %s filter named %s.", i, filter.getClass().getSimpleName())); + if (LOGGER.isDebugEnabled()) { + LOGGER.debug(String.format("Whole filter list is: %s", filters.stream().map(tmpFilter -> tmpFilter.getClass().getSimpleName()).collect(Collectors.toList()))); + } + throw filterThrowable; + } + } + }); + + return asyncResult; + } + + @Override + public Class getInterface() { + return filterInvoker.getInterface(); + } + + @Override + public URL getUrl() { + return filterInvoker.getUrl(); + } + + @Override + public boolean isAvailable() { + return filterInvoker.isAvailable(); + } + + @Override + public void destroy() { + filterInvoker.destroy(); + } + } + + class ClusterCallbackRegistrationInvoker extends CallbackRegistrationInvoker + implements ClusterInvoker { + private ClusterInvoker originalInvoker; + + public ClusterCallbackRegistrationInvoker(ClusterInvoker originalInvoker, Invoker filterInvoker, List filters) { + super(filterInvoker, filters); + this.originalInvoker = originalInvoker; + } + + public ClusterInvoker getOriginalInvoker() { + return originalInvoker; + } + + @Override + public URL getRegistryUrl() { + return getOriginalInvoker().getRegistryUrl(); + } + + @Override + public Directory getDirectory() { + return getOriginalInvoker().getDirectory(); + } + + @Override + public boolean isDestroyed() { + return getOriginalInvoker().isDestroyed(); + } + } + + + @Experimental("Works for the same purpose as FilterChainNode, replace FilterChainNode with this one when proved stable enough") + class CopyOfFilterChainNode, FILTER extends BaseFilter> implements Invoker { + TYPE originalInvoker; + Invoker nextNode; + FILTER filter; + + public CopyOfFilterChainNode(TYPE originalInvoker, Invoker nextNode, FILTER filter) { + this.originalInvoker = originalInvoker; + this.nextNode = nextNode; + this.filter = filter; + } + + public TYPE getOriginalInvoker() { + return originalInvoker; + } + + @Override + public Class getInterface() { + return originalInvoker.getInterface(); + } + + @Override + public URL getUrl() { + return originalInvoker.getUrl(); + } + + @Override + public boolean isAvailable() { + return originalInvoker.isAvailable(); + } + + @Override + public Result invoke(Invocation invocation) throws RpcException { + Result asyncResult; + try { + asyncResult = filter.invoke(nextNode, invocation); + } catch (Exception e) { + if (filter instanceof ListenableFilter) { + ListenableFilter listenableFilter = ((ListenableFilter) filter); + try { + Filter.Listener listener = listenableFilter.listener(invocation); + if (listener != null) { + listener.onError(e, originalInvoker, invocation); + } + } finally { + listenableFilter.removeListener(invocation); + } + } else if (filter instanceof FILTER.Listener) { + FILTER.Listener listener = (FILTER.Listener) filter; + listener.onError(e, originalInvoker, invocation); + } + throw e; + } finally { + + } + return asyncResult; + } + + @Override + public void destroy() { + originalInvoker.destroy(); + } + + @Override + public String toString() { + return originalInvoker.toString(); + } + } + + @Experimental("Works for the same purpose as ClusterFilterChainNode, replace ClusterFilterChainNode with this one when proved stable enough") + class CopyOfClusterFilterChainNode, FILTER extends BaseFilter> + extends CopyOfFilterChainNode implements ClusterInvoker { + public CopyOfClusterFilterChainNode(TYPE originalInvoker, Invoker nextNode, FILTER filter) { + super(originalInvoker, nextNode, filter); + } + + @Override public URL getRegistryUrl() { return getOriginalInvoker().getRegistryUrl(); diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/filter/support/ConsumerContextFilter.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/filter/support/ConsumerContextFilter.java index 846a7409c2..79eb7b8f3a 100644 --- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/filter/support/ConsumerContextFilter.java +++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/filter/support/ConsumerContextFilter.java @@ -93,35 +93,40 @@ public class ConsumerContextFilter implements ClusterFilter, ClusterFilter.Liste ((RpcInvocation) invocation).addObjectAttachments(contextAttachments); } - try { - // pass default timeout set by end user (ReferenceConfig) - Object countDown = context.getObjectAttachment(TIME_COUNTDOWN_KEY); - if (countDown != null) { - TimeoutCountDown timeoutCountDown = (TimeoutCountDown) countDown; - if (timeoutCountDown.isExpired()) { - return AsyncRpcResult.newDefaultAsyncResult(new RpcException(RpcException.TIMEOUT_TERMINATE, - "No time left for making the following call: " + invocation.getServiceName() + "." - + invocation.getMethodName() + ", terminate directly."), invocation); - } + // pass default timeout set by end user (ReferenceConfig) + Object countDown = context.getObjectAttachment(TIME_COUNTDOWN_KEY); + if (countDown != null) { + TimeoutCountDown timeoutCountDown = (TimeoutCountDown) countDown; + if (timeoutCountDown.isExpired()) { + return AsyncRpcResult.newDefaultAsyncResult(new RpcException(RpcException.TIMEOUT_TERMINATE, + "No time left for making the following call: " + invocation.getServiceName() + "." + + invocation.getMethodName() + ", terminate directly."), invocation); } - - RpcContext.removeServerContext(); - return invoker.invoke(invocation); - } finally { - RpcContext.removeServiceContext(); - RpcContext.removeClientAttachment(); } + + RpcContext.removeServerContext(); + return invoker.invoke(invocation); } @Override public void onResponse(Result appResponse, Invoker invoker, Invocation invocation) { // pass attachments to result RpcContext.getServerContext().setObjectAttachments(appResponse.getObjectAttachments()); + + removeContext(); } @Override public void onError(Throwable t, Invoker invoker, Invocation invocation) { + removeContext(); + } + private void removeContext() { + RpcContext.removeServiceContext(); + RpcContext.removeClientAttachment(); + // server context must not be removed because user might use it on callback. + // So the clear of is delayed til the start of the next rpc call, see RpcContext.removeServerContext(); in invoke() above + // RpcContext.removeServerContext(); } } diff --git a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/filter/DefaultFilterChainBuilderTest.java b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/filter/DefaultFilterChainBuilderTest.java index 1bf0b932ce..7a5779aa2c 100644 --- a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/filter/DefaultFilterChainBuilderTest.java +++ b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/filter/DefaultFilterChainBuilderTest.java @@ -19,15 +19,17 @@ package org.apache.dubbo.rpc.cluster.filter; import org.apache.dubbo.common.URL; import org.apache.dubbo.rpc.Invocation; -import org.apache.dubbo.rpc.Result; import org.apache.dubbo.rpc.Invoker; +import org.apache.dubbo.rpc.Result; +import org.apache.dubbo.rpc.model.ApplicationModel; import org.apache.dubbo.rpc.protocol.AbstractInvoker; + import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import static org.apache.dubbo.common.constants.CommonConstants.CONSUMER; import static org.apache.dubbo.common.constants.CommonConstants.INTERFACE_KEY; import static org.apache.dubbo.common.constants.CommonConstants.REFERENCE_FILTER_KEY; -import static org.apache.dubbo.common.constants.CommonConstants.CONSUMER; public class DefaultFilterChainBuilderTest { @@ -38,6 +40,7 @@ public class DefaultFilterChainBuilderTest { // verify that no filter is built by default URL urlWithoutFilter = URL.valueOf("injvm://127.0.0.1/DemoService") .addParameter(INTERFACE_KEY, DemoService.class.getName()); + urlWithoutFilter = urlWithoutFilter.setScopeModel(ApplicationModel.defaultModel()); AbstractInvoker invokerWithoutFilter = new AbstractInvoker(DemoService.class, urlWithoutFilter) { @Override protected Result doInvoke(Invocation invocation) throws Throwable { @@ -52,6 +55,7 @@ public class DefaultFilterChainBuilderTest { URL urlWithFilter = URL.valueOf("injvm://127.0.0.1/DemoService") .addParameter(INTERFACE_KEY, DemoService.class.getName()) .addParameter(REFERENCE_FILTER_KEY, "log"); + urlWithFilter = urlWithFilter.setScopeModel(ApplicationModel.defaultModel()); AbstractInvoker invokerWithFilter = new AbstractInvoker(DemoService.class, urlWithFilter) { @Override protected Result doInvoke(Invocation invocation) throws Throwable { @@ -59,8 +63,8 @@ public class DefaultFilterChainBuilderTest { } }; invokerAfterBuild = defaultFilterChainBuilder.buildInvokerChain(invokerWithFilter, REFERENCE_FILTER_KEY, CONSUMER); - Assertions.assertTrue(invokerAfterBuild instanceof FilterChainBuilder.FilterChainNode); - Assertions.assertTrue(((FilterChainBuilder.FilterChainNode) invokerAfterBuild).filter instanceof LogFilter); + Assertions.assertTrue(invokerAfterBuild instanceof FilterChainBuilder.CallbackRegistrationInvoker); + Assertions.assertEquals(1, ((FilterChainBuilder.CallbackRegistrationInvoker) invokerAfterBuild).filters.size()); } @@ -71,6 +75,7 @@ public class DefaultFilterChainBuilderTest { // verify that no filter is built by default URL urlWithoutFilter = URL.valueOf("dubbo://127.0.0.1:20880/DemoService") .addParameter(INTERFACE_KEY, DemoService.class.getName()); + urlWithoutFilter = urlWithoutFilter.setScopeModel(ApplicationModel.defaultModel()); AbstractInvoker invokerWithoutFilter = new AbstractInvoker(DemoService.class, urlWithoutFilter) { @Override protected Result doInvoke(Invocation invocation) throws Throwable { @@ -85,6 +90,7 @@ public class DefaultFilterChainBuilderTest { URL urlWithFilter = URL.valueOf("dubbo://127.0.0.1:20880/DemoService") .addParameter(INTERFACE_KEY, DemoService.class.getName()) .addParameter(REFERENCE_FILTER_KEY, "log"); + urlWithFilter = urlWithFilter.setScopeModel(ApplicationModel.defaultModel()); AbstractInvoker invokerWithFilter = new AbstractInvoker(DemoService.class, urlWithFilter) { @Override protected Result doInvoke(Invocation invocation) throws Throwable { @@ -92,8 +98,7 @@ public class DefaultFilterChainBuilderTest { } }; invokerAfterBuild = defaultFilterChainBuilder.buildInvokerChain(invokerWithFilter, REFERENCE_FILTER_KEY, CONSUMER); - Assertions.assertTrue(invokerAfterBuild instanceof FilterChainBuilder.FilterChainNode); - Assertions.assertTrue(((FilterChainBuilder.FilterChainNode) invokerAfterBuild).filter instanceof LogFilter); + Assertions.assertTrue(invokerAfterBuild instanceof FilterChainBuilder.CallbackRegistrationInvoker); } } diff --git a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/FailoverClusterInvokerTest.java b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/FailoverClusterInvokerTest.java index 4b9fba5b9d..1fd407508e 100644 --- a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/FailoverClusterInvokerTest.java +++ b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/FailoverClusterInvokerTest.java @@ -295,7 +295,7 @@ public class FailoverClusterInvokerTest { } invokers.clear(); MockInvoker invoker3 = new MockInvoker<>(Demo.class, url); - invoker3.setResult(AsyncRpcResult.newDefaultAsyncResult(null)); + invoker3.setResult(AsyncRpcResult.newDefaultAsyncResult(mock(RpcInvocation.class))); invokers.add(invoker3); dic.notify(invokers); return null; diff --git a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/MergeableClusterInvokerTest.java b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/MergeableClusterInvokerTest.java index bb2ff1b855..b95d0ef1a5 100644 --- a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/MergeableClusterInvokerTest.java +++ b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/MergeableClusterInvokerTest.java @@ -23,6 +23,7 @@ import org.apache.dubbo.rpc.Invocation; import org.apache.dubbo.rpc.Invoker; import org.apache.dubbo.rpc.Result; import org.apache.dubbo.rpc.RpcException; +import org.apache.dubbo.rpc.RpcInvocation; import org.apache.dubbo.rpc.cluster.Directory; import org.apache.dubbo.rpc.model.ApplicationModel; import org.apache.dubbo.rpc.model.ModuleModel; @@ -41,8 +42,6 @@ import java.util.Map; import static org.apache.dubbo.common.constants.CommonConstants.GROUP_KEY; import static org.apache.dubbo.rpc.Constants.MERGER_KEY; -import static org.hamcrest.CoreMatchers.is; -import static org.hamcrest.MatcherAssert.assertThat; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -55,7 +54,7 @@ public class MergeableClusterInvokerTest { private Directory directory = mock(Directory.class); private Invoker firstInvoker = mock(Invoker.class); private Invoker secondInvoker = mock(Invoker.class); - private Invocation invocation = mock(Invocation.class); + private Invocation invocation = mock(RpcInvocation.class); private ModuleModel moduleModel = mock(ModuleModel.class); private MergeableClusterInvoker mergeableClusterInvoker; @@ -98,7 +97,7 @@ public class MergeableClusterInvokerTest { directory = mock(Directory.class); firstInvoker = mock(Invoker.class); secondInvoker = mock(Invoker.class); - invocation = mock(Invocation.class); + invocation = mock(RpcInvocation.class); } diff --git a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/wrapper/AbstractClusterTest.java b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/wrapper/AbstractClusterTest.java index 9eccf0a291..6245b71cf4 100644 --- a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/wrapper/AbstractClusterTest.java +++ b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/wrapper/AbstractClusterTest.java @@ -28,6 +28,7 @@ import org.apache.dubbo.rpc.cluster.LoadBalance; import org.apache.dubbo.rpc.cluster.filter.DemoService; import org.apache.dubbo.rpc.cluster.filter.FilterChainBuilder; import org.apache.dubbo.rpc.cluster.support.AbstractClusterInvoker; +import org.apache.dubbo.rpc.model.ApplicationModel; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; @@ -55,13 +56,12 @@ public class AbstractClusterTest { 2181, "org.apache.dubbo.registry.RegistryService", parameters); - URL consumerUrl = new ServiceConfigURL("dubbo", "127.0.0.1", 20881, DemoService.class.getName(), parameters); - + consumerUrl = consumerUrl.setScopeModel(ApplicationModel.defaultModel().getInternalModule()); Directory directory = mock(Directory.class); when(directory.getUrl()).thenReturn(url); when(directory.getConsumerUrl()).thenReturn(consumerUrl); @@ -69,7 +69,7 @@ public class AbstractClusterTest { Invoker invoker = demoCluster.join(directory, true); Assertions.assertTrue(invoker instanceof AbstractCluster.ClusterFilterInvoker); Assertions.assertTrue(((AbstractCluster.ClusterFilterInvoker) invoker).getFilterInvoker() - instanceof FilterChainBuilder.ClusterFilterChainNode); + instanceof FilterChainBuilder.ClusterCallbackRegistrationInvoker); } diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/Experimental.java b/dubbo-common/src/main/java/org/apache/dubbo/common/Experimental.java index 64f4cd3eff..746bc1b7ad 100644 --- a/dubbo-common/src/main/java/org/apache/dubbo/common/Experimental.java +++ b/dubbo-common/src/main/java/org/apache/dubbo/common/Experimental.java @@ -22,7 +22,7 @@ import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target; /** - * Indicating unstable API, may get removed or changed in the next release. + * Indicating unstable API, may get removed or changed in future releases. */ @Retention(RetentionPolicy.CLASS) @Target({ diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/config/CompositeConfiguration.java b/dubbo-common/src/main/java/org/apache/dubbo/common/config/CompositeConfiguration.java index 887591bd77..a1e5e0e337 100644 --- a/dubbo-common/src/main/java/org/apache/dubbo/common/config/CompositeConfiguration.java +++ b/dubbo-common/src/main/java/org/apache/dubbo/common/config/CompositeConfiguration.java @@ -21,19 +21,19 @@ import org.apache.dubbo.common.logger.LoggerFactory; import org.apache.dubbo.common.utils.ArrayUtils; import java.util.Arrays; -import java.util.LinkedList; import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; /** * This is an abstraction specially customized for the sequence Dubbo retrieves properties. */ public class CompositeConfiguration implements Configuration { - private Logger logger = LoggerFactory.getLogger(CompositeConfiguration.class); + private final Logger logger = LoggerFactory.getLogger(CompositeConfiguration.class); /** * List holding all the configuration */ - private List configList = new LinkedList<>(); + private final List configList = new CopyOnWriteArrayList<>(); //FIXME, consider change configList to SortedMap to replace this boolean status. private boolean dynamicIncluded; @@ -47,15 +47,15 @@ public class CompositeConfiguration implements Configuration { } } - public void setDynamicIncluded(boolean dynamicIncluded) { - this.dynamicIncluded = dynamicIncluded; - } - //FIXME, consider change configList to SortedMap to replace this boolean status. public boolean isDynamicIncluded() { return dynamicIncluded; } + public void setDynamicIncluded(boolean dynamicIncluded) { + this.dynamicIncluded = dynamicIncluded; + } + public void addConfiguration(Configuration configuration) { if (configList.contains(configuration)) { return; @@ -80,7 +80,8 @@ public class CompositeConfiguration implements Configuration { return value; } } catch (Exception e) { - logger.error("Error when trying to get value for key " + key + " from " + config + ", will continue to try the next one."); + logger.error("Error when trying to get value for key " + key + " from " + config + ", " + + "will continue to try the next one."); } } return null; diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/config/ConfigurationCache.java b/dubbo-common/src/main/java/org/apache/dubbo/common/config/ConfigurationCache.java index 977cdfb145..c4e5ab3e71 100644 --- a/dubbo-common/src/main/java/org/apache/dubbo/common/config/ConfigurationCache.java +++ b/dubbo-common/src/main/java/org/apache/dubbo/common/config/ConfigurationCache.java @@ -16,6 +16,7 @@ */ package org.apache.dubbo.common.config; +import org.apache.dubbo.common.utils.StringUtils; import org.apache.dubbo.rpc.model.ScopeModel; import java.util.Map; @@ -37,7 +38,7 @@ public class ConfigurationCache { */ public String computeIfAbsent(String key, Function function) { String value = cache.get(key); - if (value == null) { + if (StringUtils.isEmpty(value)) { // lock free, tolerate repeat apply, will return previous value cache.putIfAbsent(key, function.apply(key)); value = cache.get(key); diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/config/ConfigurationUtils.java b/dubbo-common/src/main/java/org/apache/dubbo/common/config/ConfigurationUtils.java index fd25f030f5..d0c2f2314f 100644 --- a/dubbo-common/src/main/java/org/apache/dubbo/common/config/ConfigurationUtils.java +++ b/dubbo-common/src/main/java/org/apache/dubbo/common/config/ConfigurationUtils.java @@ -21,6 +21,7 @@ import org.apache.dubbo.common.extension.ExtensionAccessor; import org.apache.dubbo.common.extension.ExtensionLoader; import org.apache.dubbo.common.logger.Logger; import org.apache.dubbo.common.logger.LoggerFactory; +import org.apache.dubbo.common.utils.CollectionUtils; import org.apache.dubbo.common.utils.StringUtils; import org.apache.dubbo.rpc.model.ApplicationModel; import org.apache.dubbo.rpc.model.ScopeModel; @@ -35,6 +36,7 @@ import java.util.LinkedHashSet; import java.util.LinkedList; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Properties; import java.util.Set; @@ -228,7 +230,7 @@ public class ConfigurationUtils { resultMap = new LinkedHashMap<>(); } - if (null != configMap) { + if (CollectionUtils.isNotEmptyMap(configMap)) { for(Map.Entry entry : configMap.entrySet()) { String key = entry.getKey(); V val = entry.getValue(); @@ -238,8 +240,11 @@ public class ConfigurationUtils { String k = key.substring(prefix.length()); // convert camelCase/snake_case to kebab-case - k = StringUtils.convertToSplitName(k, "-"); - resultMap.putIfAbsent(k, val); + String newK = StringUtils.convertToSplitName(k, "-"); + resultMap.putIfAbsent(newK, val); + if (!Objects.equals(k, newK)) { + resultMap.putIfAbsent(k, val); + } } } } diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/config/InmemoryConfiguration.java b/dubbo-common/src/main/java/org/apache/dubbo/common/config/InmemoryConfiguration.java index 92d808de9f..7b3fa985c7 100644 --- a/dubbo-common/src/main/java/org/apache/dubbo/common/config/InmemoryConfiguration.java +++ b/dubbo-common/src/main/java/org/apache/dubbo/common/config/InmemoryConfiguration.java @@ -16,6 +16,8 @@ */ package org.apache.dubbo.common.config; +import org.apache.dubbo.common.utils.CollectionUtils; + import java.util.LinkedHashMap; import java.util.Map; @@ -56,7 +58,7 @@ public class InmemoryConfiguration implements Configuration { * Add a set of properties into the store */ public void addProperties(Map properties) { - if (properties != null) { + if (CollectionUtils.isNotEmptyMap(properties)) { this.store.putAll(properties); } } @@ -65,7 +67,7 @@ public class InmemoryConfiguration implements Configuration { * set store */ public void setProperties(Map properties) { - if (properties != null) { + if (CollectionUtils.isNotEmptyMap(properties)) { this.store = properties; } } diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java b/dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java index 7cf182f589..bb3833b5b1 100644 --- a/dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java +++ b/dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java @@ -507,7 +507,17 @@ public interface CommonConstants { String DUBBO_INTERNAL_APPLICATION = "DUBBO_INTERNAL_APPLICATION"; String RETRY_TIMES_KEY = "retry-times"; + String RETRY_PERIOD_KEY = "retry-period"; + String SYNC_REPORT_KEY = "sync-report"; + String CYCLE_REPORT_KEY = "cycle-report"; + + String WORKING_CLASSLOADER_KEY = "WORKING_CLASSLOADER"; + + String STAGED_CLASSLOADER_KEY = "STAGED_CLASSLOADER"; + + String PROVIDER_ASYNC_KEY = "PROVIDER_ASYNC"; + } diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/constants/RegistryConstants.java b/dubbo-common/src/main/java/org/apache/dubbo/common/constants/RegistryConstants.java index 6029293bad..d8061cbec1 100644 --- a/dubbo-common/src/main/java/org/apache/dubbo/common/constants/RegistryConstants.java +++ b/dubbo-common/src/main/java/org/apache/dubbo/common/constants/RegistryConstants.java @@ -61,10 +61,6 @@ public interface RegistryConstants { String COMPATIBLE_CONFIG_KEY = "compatible_config"; - String REGISTRY_PUBLISH_INTERFACE_KEY = "publish-interface"; - - String REGISTRY_PUBLISH_INSTANCE_KEY = "publish-instance"; - String REGISTER_MODE_KEY = "register-mode"; String DUBBO_REGISTER_MODE_DEFAULT_KEY = "dubbo.application.register-mode"; @@ -132,4 +128,6 @@ public interface RegistryConstants { String INIT = "INIT"; float DEFAULT_HASHMAP_LOAD_FACTOR = 0.75f; + + String ENABLE_EMPTY_PROTECTION_KEY = "enable-empty-protection"; } diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/extension/ExtensionDirector.java b/dubbo-common/src/main/java/org/apache/dubbo/common/extension/ExtensionDirector.java index 578fcf9458..1614748ceb 100644 --- a/dubbo-common/src/main/java/org/apache/dubbo/common/extension/ExtensionDirector.java +++ b/dubbo-common/src/main/java/org/apache/dubbo/common/extension/ExtensionDirector.java @@ -35,11 +35,11 @@ public class ExtensionDirector implements ExtensionAccessor { private final ConcurrentMap, ExtensionLoader> extensionLoadersMap = new ConcurrentHashMap<>(64); private final ConcurrentMap, ExtensionScope> extensionScopeMap = new ConcurrentHashMap<>(64); - private ExtensionDirector parent; + private final ExtensionDirector parent; private final ExtensionScope scope; - private List extensionPostProcessors = new ArrayList<>(); - private ScopeModel scopeModel; - private AtomicBoolean destroyed = new AtomicBoolean(); + private final List extensionPostProcessors = new ArrayList<>(); + private final ScopeModel scopeModel; + private final AtomicBoolean destroyed = new AtomicBoolean(); public ExtensionDirector(ExtensionDirector parent, ExtensionScope scope, ScopeModel scopeModel) { this.parent = parent; diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/extension/support/ActivateComparator.java b/dubbo-common/src/main/java/org/apache/dubbo/common/extension/support/ActivateComparator.java index 5c755ae1df..1b08ec2173 100644 --- a/dubbo-common/src/main/java/org/apache/dubbo/common/extension/support/ActivateComparator.java +++ b/dubbo-common/src/main/java/org/apache/dubbo/common/extension/support/ActivateComparator.java @@ -32,7 +32,7 @@ import java.util.concurrent.ConcurrentHashMap; */ public class ActivateComparator implements Comparator> { - private ExtensionDirector extensionDirector; + private final ExtensionDirector extensionDirector; private final Map, ActivateInfo> activateInfoMap = new ConcurrentHashMap<>(); public ActivateComparator(ExtensionDirector extensionDirector) { @@ -128,9 +128,9 @@ public class ActivateComparator implements Comparator> { info.before = activate.before(); info.after = activate.after(); info.order = activate.order(); - } else if (clazz.isAnnotationPresent(com.alibaba.dubbo.common.extension.Activate.class)){ + } else if (clazz.isAnnotationPresent(com.alibaba.dubbo.common.extension.Activate.class)) { com.alibaba.dubbo.common.extension.Activate activate = clazz.getAnnotation( - com.alibaba.dubbo.common.extension.Activate.class); + com.alibaba.dubbo.common.extension.Activate.class); info.before = activate.before(); info.after = activate.after(); info.order = activate.order(); diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/extension/support/MultiInstanceActivateComparator.java b/dubbo-common/src/main/java/org/apache/dubbo/common/extension/support/MultiInstanceActivateComparator.java new file mode 100644 index 0000000000..5af8ed5f19 --- /dev/null +++ b/dubbo-common/src/main/java/org/apache/dubbo/common/extension/support/MultiInstanceActivateComparator.java @@ -0,0 +1,176 @@ +/* + * 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.common.extension.support; + +import org.apache.dubbo.common.extension.Activate; +import org.apache.dubbo.common.extension.ExtensionDirector; +import org.apache.dubbo.common.extension.ExtensionLoader; +import org.apache.dubbo.common.extension.SPI; +import org.apache.dubbo.common.utils.ArrayUtils; + +import java.util.List; +import java.util.Comparator; +import java.util.Map; +import java.util.Arrays; +import java.util.concurrent.ConcurrentHashMap; + +public class MultiInstanceActivateComparator implements Comparator> { + + private final List extensionDirectors; + private final Map, ActivateInfo> activateInfoMap = new ConcurrentHashMap<>(); + + public MultiInstanceActivateComparator(List extensionDirectors) { + this.extensionDirectors = extensionDirectors; + } + + @Override + public int compare(Class o1, Class o2) { + if (o1 == null && o2 == null) { + return 0; + } + if (o1 == null) { + return -1; + } + if (o2 == null) { + return 1; + } + if (o1.equals(o2)) { + return 0; + } + + Class inf = findSpi(o1); + + ActivateInfo a1 = parseActivate(o1); + ActivateInfo a2 = parseActivate(o2); + + if ((a1.applicableToCompare() || a2.applicableToCompare()) && inf != null) { + + + if (a1.applicableToCompare()) { + String n2 = null; + for (ExtensionDirector director : extensionDirectors) { + ExtensionLoader extensionLoader = director.getExtensionLoader(inf); + n2 = extensionLoader.getExtensionName(o2); + if (n2 != null) { + break; + } + } + if (a1.isLess(n2)) { + return -1; + } + + if (a1.isMore(n2)) { + return 1; + } + } + + if (a2.applicableToCompare()) { + String n1 = null; + for (ExtensionDirector director : extensionDirectors) { + ExtensionLoader extensionLoader = director.getExtensionLoader(inf); + n1 = extensionLoader.getExtensionName(o1); + if (n1 != null) { + break; + } + } + + if (a2.isLess(n1)) { + return 1; + } + + if (a2.isMore(n1)) { + return -1; + } + } + + return a1.order > a2.order ? 1 : -1; + } + + // In order to avoid the problem of inconsistency between the loading order of two filters + // in different loading scenarios without specifying the order attribute of the filter, + // when the order is the same, compare its filterName + if (a1.order > a2.order) { + return 1; + } else if (a1.order == a2.order) { + return o1.getSimpleName().compareTo(o2.getSimpleName()) > 0 ? 1 : -1; + } else { + return -1; + } + } + + private Class findSpi(Class clazz) { + if (clazz.getInterfaces().length == 0) { + return null; + } + + for (Class intf : clazz.getInterfaces()) { + if (intf.isAnnotationPresent(SPI.class)) { + return intf; + } else { + Class result = findSpi(intf); + if (result != null) { + return result; + } + } + } + + return null; + } + + private ActivateInfo parseActivate(Class clazz) { + ActivateInfo info = activateInfoMap.get(clazz); + if (info != null) { + return info; + } + info = new ActivateInfo(); + if (clazz.isAnnotationPresent(Activate.class)) { + Activate activate = clazz.getAnnotation(Activate.class); + info.before = activate.before(); + info.after = activate.after(); + info.order = activate.order(); + } else if (clazz.isAnnotationPresent(com.alibaba.dubbo.common.extension.Activate.class)) { + com.alibaba.dubbo.common.extension.Activate activate = clazz.getAnnotation( + com.alibaba.dubbo.common.extension.Activate.class); + info.before = activate.before(); + info.after = activate.after(); + info.order = activate.order(); + } else { + info.order = 0; + } + activateInfoMap.put(clazz, info); + return info; + } + + private static class ActivateInfo { + private String[] before; + private String[] after; + private int order; + + private boolean applicableToCompare() { + return ArrayUtils.isNotEmpty(before) || ArrayUtils.isNotEmpty(after); + } + + private boolean isLess(String name) { + return Arrays.asList(before).contains(name); + } + + private boolean isMore(String name) { + return Arrays.asList(after).contains(name); + } + } +} diff --git a/dubbo-common/src/main/java/org/apache/dubbo/config/ApplicationConfig.java b/dubbo-common/src/main/java/org/apache/dubbo/config/ApplicationConfig.java index 772c313adb..97f6e16505 100644 --- a/dubbo-common/src/main/java/org/apache/dubbo/config/ApplicationConfig.java +++ b/dubbo-common/src/main/java/org/apache/dubbo/config/ApplicationConfig.java @@ -51,8 +51,8 @@ import static org.apache.dubbo.common.constants.QosConstants.ACCEPT_FOREIGN_IP; import static org.apache.dubbo.common.constants.QosConstants.QOS_ENABLE; import static org.apache.dubbo.common.constants.QosConstants.QOS_HOST; import static org.apache.dubbo.common.constants.QosConstants.QOS_PORT; -import static org.apache.dubbo.common.constants.RegistryConstants.REGISTRY_PUBLISH_INSTANCE_KEY; -import static org.apache.dubbo.common.constants.RegistryConstants.REGISTRY_PUBLISH_INTERFACE_KEY; +import static org.apache.dubbo.common.constants.RegistryConstants.ENABLE_EMPTY_PROTECTION_KEY; +import static org.apache.dubbo.common.constants.RegistryConstants.REGISTER_MODE_KEY; import static org.apache.dubbo.config.Constants.DEVELOPMENT_ENVIRONMENT; import static org.apache.dubbo.config.Constants.PRODUCTION_ENVIRONMENT; import static org.apache.dubbo.config.Constants.TEST_ENVIRONMENT; @@ -167,10 +167,6 @@ public class ApplicationConfig extends AbstractConfig { private Boolean enableFileCache; - private Boolean publishInterface; - - private Boolean publishInstance; - /** * The preferred protocol(name) of this application * convenient for places where it's hard to determine which is the preferred protocol @@ -196,6 +192,10 @@ public class ApplicationConfig extends AbstractConfig { private String startupProbe; + private String registerMode; + + private Boolean enableEmptyProtection; + public ApplicationConfig() { } @@ -497,22 +497,22 @@ public class ApplicationConfig extends AbstractConfig { this.enableFileCache = enableFileCache; } - @Parameter(key = REGISTRY_PUBLISH_INTERFACE_KEY) - public Boolean getPublishInterface() { - return publishInterface; + @Parameter(key = REGISTER_MODE_KEY) + public String getRegisterMode() { + return registerMode; } - public void setPublishInterface(Boolean publishInterface) { - this.publishInterface = publishInterface; + public void setRegisterMode(String registerMode) { + this.registerMode = registerMode; } - @Parameter(key = REGISTRY_PUBLISH_INSTANCE_KEY) - public Boolean getPublishInstance() { - return publishInstance; + @Parameter(key = ENABLE_EMPTY_PROTECTION_KEY) + public Boolean getEnableEmptyProtection() { + return enableEmptyProtection; } - public void setPublishInstance(Boolean publishInstance) { - this.publishInstance = publishInstance; + public void setEnableEmptyProtection(Boolean enableEmptyProtection) { + this.enableEmptyProtection = enableEmptyProtection; } @Parameter(excluded = true, key = APPLICATION_PROTOCOL_KEY) diff --git a/dubbo-common/src/main/java/org/apache/dubbo/config/RegistryConfig.java b/dubbo-common/src/main/java/org/apache/dubbo/config/RegistryConfig.java index d3d99f100b..f3863a589a 100644 --- a/dubbo-common/src/main/java/org/apache/dubbo/config/RegistryConfig.java +++ b/dubbo-common/src/main/java/org/apache/dubbo/config/RegistryConfig.java @@ -26,9 +26,9 @@ import java.util.Map; import static org.apache.dubbo.common.constants.CommonConstants.EXTRA_KEYS_KEY; import static org.apache.dubbo.common.constants.CommonConstants.SHUTDOWN_WAIT_KEY; +import static org.apache.dubbo.common.constants.RegistryConstants.ENABLE_EMPTY_PROTECTION_KEY; +import static org.apache.dubbo.common.constants.RegistryConstants.REGISTER_MODE_KEY; import static org.apache.dubbo.common.constants.RegistryConstants.REGISTRY_CLUSTER_KEY; -import static org.apache.dubbo.common.constants.RegistryConstants.REGISTRY_PUBLISH_INSTANCE_KEY; -import static org.apache.dubbo.common.constants.RegistryConstants.REGISTRY_PUBLISH_INTERFACE_KEY; import static org.apache.dubbo.common.constants.RemotingConstants.BACKUP_KEY; import static org.apache.dubbo.common.utils.PojoUtils.updatePropertyIfAbsent; @@ -180,9 +180,9 @@ public class RegistryConfig extends AbstractConfig { */ private Integer weight; - private Boolean publishInterface; + private String registerMode; - private Boolean publishInstance; + private Boolean enableEmptyProtection; public RegistryConfig() { } @@ -519,22 +519,22 @@ public class RegistryConfig extends AbstractConfig { this.weight = weight; } - @Parameter(key = REGISTRY_PUBLISH_INTERFACE_KEY) - public Boolean getPublishInterface() { - return publishInterface; + @Parameter(key = REGISTER_MODE_KEY) + public String getRegisterMode() { + return registerMode; } - public void setPublishInterface(Boolean publishInterface) { - this.publishInterface = publishInterface; + public void setRegisterMode(String registerMode) { + this.registerMode = registerMode; } - @Parameter(key = REGISTRY_PUBLISH_INSTANCE_KEY) - public Boolean getPublishInstance() { - return publishInstance; + @Parameter(key = ENABLE_EMPTY_PROTECTION_KEY) + public Boolean getEnableEmptyProtection() { + return enableEmptyProtection; } - public void setPublishInstance(Boolean publishInstance) { - this.publishInstance = publishInstance; + public void setEnableEmptyProtection(Boolean enableEmptyProtection) { + this.enableEmptyProtection = enableEmptyProtection; } @Override diff --git a/dubbo-common/src/main/java/org/apache/dubbo/rpc/model/ScopeModelUtil.java b/dubbo-common/src/main/java/org/apache/dubbo/rpc/model/ScopeModelUtil.java index 2d3e06a944..5e18241594 100644 --- a/dubbo-common/src/main/java/org/apache/dubbo/rpc/model/ScopeModelUtil.java +++ b/dubbo-common/src/main/java/org/apache/dubbo/rpc/model/ScopeModelUtil.java @@ -28,7 +28,7 @@ public class ScopeModelUtil { return getDefaultScopeModel(type); } - private static ScopeModel getDefaultScopeModel(Class type) { + private static ScopeModel getDefaultScopeModel(Class type) { SPI spi = type.getAnnotation(SPI.class); if (spi == null) { throw new IllegalArgumentException("SPI annotation not found for class: " + type.getName()); @@ -40,8 +40,9 @@ public class ScopeModelUtil { return ApplicationModel.defaultModel(); case MODULE: return ApplicationModel.defaultModel().getDefaultModule(); + default: + throw new IllegalStateException("Unable to get default scope model for type: " + type.getName()); } - throw new IllegalStateException("Unable to get default scope model for type: " + type.getName()); } public static ModuleModel getModuleModel(ScopeModel scopeModel) { diff --git a/dubbo-common/src/test/java/org/apache/dubbo/rpc/model/FrameworkModelTest.java b/dubbo-common/src/test/java/org/apache/dubbo/rpc/model/FrameworkModelTest.java index 9967d57745..ee32037ff8 100644 --- a/dubbo-common/src/test/java/org/apache/dubbo/rpc/model/FrameworkModelTest.java +++ b/dubbo-common/src/test/java/org/apache/dubbo/rpc/model/FrameworkModelTest.java @@ -27,6 +27,7 @@ import org.junit.jupiter.api.Test; public class FrameworkModelTest { @Test public void testInitialize() { + FrameworkModel.destroyAll(); FrameworkModel frameworkModel = new FrameworkModel(); Assertions.assertNull(frameworkModel.getParent()); diff --git a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/deploy/DefaultModuleDeployer.java b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/deploy/DefaultModuleDeployer.java index 173b343e79..b2c4bbff22 100644 --- a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/deploy/DefaultModuleDeployer.java +++ b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/deploy/DefaultModuleDeployer.java @@ -385,8 +385,9 @@ public class DefaultModuleDeployer extends AbstractDeployer impleme } } } catch (Throwable t) { - logger.error(getIdentifier() + " refer catch error", t); + logger.error(getIdentifier() + " refer catch error."); referenceCache.destroy(rc); + throw t; } }); } diff --git a/dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/ReferenceConfigTest.java b/dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/ReferenceConfigTest.java index 904ce83429..c279eeea1e 100644 --- a/dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/ReferenceConfigTest.java +++ b/dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/ReferenceConfigTest.java @@ -110,8 +110,6 @@ import static org.apache.dubbo.common.constants.QosConstants.ACCEPT_FOREIGN_IP; import static org.apache.dubbo.common.constants.QosConstants.QOS_ENABLE; import static org.apache.dubbo.common.constants.QosConstants.QOS_HOST; import static org.apache.dubbo.common.constants.QosConstants.QOS_PORT; -import static org.apache.dubbo.common.constants.RegistryConstants.REGISTRY_PUBLISH_INSTANCE_KEY; -import static org.apache.dubbo.common.constants.RegistryConstants.REGISTRY_PUBLISH_INTERFACE_KEY; import static org.apache.dubbo.registry.Constants.REGISTER_IP_KEY; import static org.apache.dubbo.rpc.Constants.DEFAULT_STUB_EVENT; import static org.apache.dubbo.rpc.Constants.LOCAL_KEY; @@ -177,8 +175,6 @@ public class ReferenceConfigTest { applicationConfig.setRegisterConsumer(false); applicationConfig.setRepository("repository1"); applicationConfig.setEnableFileCache(false); - applicationConfig.setPublishInstance(false); - applicationConfig.setPublishInterface(false); applicationConfig.setProtocol("dubbo"); applicationConfig.setMetadataServicePort(88888); applicationConfig.setMetadataServiceProtocol("tri"); @@ -315,11 +311,6 @@ public class ReferenceConfigTest { serviceMetadata.getAttachments().get("repository")); Assertions.assertEquals(applicationConfig.getEnableFileCache().toString(), serviceMetadata.getAttachments().get(REGISTRY_LOCAL_FILE_CACHE_ENABLED)); - Assertions.assertEquals(applicationConfig.getPublishInstance().toString(), - serviceMetadata.getAttachments().get(REGISTRY_PUBLISH_INSTANCE_KEY)); - Assertions.assertEquals(applicationConfig.getPublishInterface().toString(), - serviceMetadata.getAttachments().get(REGISTRY_PUBLISH_INTERFACE_KEY)); - Assertions.assertTrue(serviceMetadata.getAttachments().containsKey(REGISTRY_PUBLISH_INTERFACE_KEY)); Assertions.assertEquals(applicationConfig.getMetadataServicePort().toString(), serviceMetadata.getAttachments().get(METADATA_SERVICE_PORT_KEY)); Assertions.assertEquals(applicationConfig.getMetadataServiceProtocol().toString(), diff --git a/dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/integration/AbstractRegistryCenterExporterListener.java b/dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/integration/AbstractRegistryCenterExporterListener.java index 69a3d5e6fd..74e1e69592 100644 --- a/dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/integration/AbstractRegistryCenterExporterListener.java +++ b/dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/integration/AbstractRegistryCenterExporterListener.java @@ -26,9 +26,9 @@ import org.apache.dubbo.rpc.listener.ListenerExporterWrapper; import java.lang.reflect.Field; import java.util.ArrayList; import java.util.Collections; +import java.util.HashSet; import java.util.List; import java.util.Set; -import java.util.HashSet; /** * The abstraction of {@link ExporterListener} is to record exported exporters, which should be extended by different sub-classes. @@ -56,12 +56,13 @@ public abstract class AbstractRegistryCenterExporterListener implements Exporter @Override public void exported(Exporter exporter) throws RpcException { ListenerExporterWrapper listenerExporterWrapper = (ListenerExporterWrapper) exporter; - FilterChainBuilder.FilterChainNode filterChainNode = (FilterChainBuilder.FilterChainNode) listenerExporterWrapper.getInvoker(); - if (filterChainNode == null || - filterChainNode.getInterface() != getInterface()) { + FilterChainBuilder.CallbackRegistrationInvoker callbackRegistrationInvoker = (FilterChainBuilder.CallbackRegistrationInvoker) listenerExporterWrapper.getInvoker(); + if (callbackRegistrationInvoker == null || + callbackRegistrationInvoker.getInterface() != getInterface()) { return; } exportedExporters.add(exporter); + FilterChainBuilder.CopyOfFilterChainNode filterChainNode = getFilterChainNode(callbackRegistrationInvoker); do { Filter filter = this.getFilter(filterChainNode); if (filter != null) { @@ -96,7 +97,24 @@ public abstract class AbstractRegistryCenterExporterListener implements Exporter /** * Use reflection to obtain {@link Filter} */ - private Filter getFilter(FilterChainBuilder.FilterChainNode filterChainNode) { + private FilterChainBuilder.CopyOfFilterChainNode getFilterChainNode(FilterChainBuilder.CallbackRegistrationInvoker callbackRegistrationInvoker) { + if (callbackRegistrationInvoker != null) { + Field field = null; + try { + field = callbackRegistrationInvoker.getClass().getDeclaredField("filterInvoker"); + field.setAccessible(true); + return (FilterChainBuilder.CopyOfFilterChainNode) field.get(callbackRegistrationInvoker); + } catch (NoSuchFieldException | IllegalAccessException e) { + // ignore + } + } + return null; + } + + /** + * Use reflection to obtain {@link Filter} + */ + private Filter getFilter(FilterChainBuilder.CopyOfFilterChainNode filterChainNode) { if (filterChainNode != null) { Field field = null; try { @@ -111,17 +129,17 @@ public abstract class AbstractRegistryCenterExporterListener implements Exporter } /** - * Use reflection to obtain {@link FilterChainBuilder.FilterChainNode} + * Use reflection to obtain {@link FilterChainBuilder.CopyOfFilterChainNode} */ - private FilterChainBuilder.FilterChainNode getNextNode(FilterChainBuilder.FilterChainNode filterChainNode) { + private FilterChainBuilder.CopyOfFilterChainNode getNextNode(FilterChainBuilder.CopyOfFilterChainNode filterChainNode) { if (filterChainNode != null) { Field field = null; try { field = filterChainNode.getClass().getDeclaredField("nextNode"); field.setAccessible(true); Object object = field.get(filterChainNode); - if (object instanceof FilterChainBuilder.FilterChainNode) { - return (FilterChainBuilder.FilterChainNode) object; + if (object instanceof FilterChainBuilder.CopyOfFilterChainNode) { + return (FilterChainBuilder.CopyOfFilterChainNode) object; } } catch (NoSuchFieldException | IllegalAccessException e) { // ignore diff --git a/dubbo-config/dubbo-config-spring/pom.xml b/dubbo-config/dubbo-config-spring/pom.xml index 648d6e3132..1baf323926 100644 --- a/dubbo-config/dubbo-config-spring/pom.xml +++ b/dubbo-config/dubbo-config-spring/pom.xml @@ -152,6 +152,11 @@ junit-vintage-engine org.junit.vintage + + + org.apache.logging.log4j + log4j-api + diff --git a/dubbo-config/dubbo-config-spring/src/main/resources/META-INF/compat/dubbo.xsd b/dubbo-config/dubbo-config-spring/src/main/resources/META-INF/compat/dubbo.xsd index ef02b50dfc..6b0b20bb90 100644 --- a/dubbo-config/dubbo-config-spring/src/main/resources/META-INF/compat/dubbo.xsd +++ b/dubbo-config/dubbo-config-spring/src/main/resources/META-INF/compat/dubbo.xsd @@ -420,10 +420,16 @@ - + - + + + + + + + @@ -639,6 +645,18 @@ + + + + + + + + + + + + diff --git a/dubbo-config/dubbo-config-spring/src/main/resources/META-INF/dubbo.xsd b/dubbo-config/dubbo-config-spring/src/main/resources/META-INF/dubbo.xsd index 8f409a6804..c6f8205f2a 100644 --- a/dubbo-config/dubbo-config-spring/src/main/resources/META-INF/dubbo.xsd +++ b/dubbo-config/dubbo-config-spring/src/main/resources/META-INF/dubbo.xsd @@ -427,10 +427,16 @@ - + - + + + + + + + @@ -674,16 +680,16 @@ - + - + - + - + diff --git a/dubbo-demo/dubbo-demo-spring-boot/dubbo-demo-spring-boot-consumer/pom.xml b/dubbo-demo/dubbo-demo-spring-boot/dubbo-demo-spring-boot-consumer/pom.xml index 24aeb0e6de..570ab67c60 100644 --- a/dubbo-demo/dubbo-demo-spring-boot/dubbo-demo-spring-boot-consumer/pom.xml +++ b/dubbo-demo/dubbo-demo-spring-boot/dubbo-demo-spring-boot-consumer/pom.xml @@ -102,6 +102,13 @@ org.springframework.boot spring-boot-starter-logging ${spring-boot.version} + + + + org.apache.logging.log4j + log4j-api + + diff --git a/dubbo-demo/dubbo-demo-spring-boot/dubbo-demo-spring-boot-provider/pom.xml b/dubbo-demo/dubbo-demo-spring-boot/dubbo-demo-spring-boot-provider/pom.xml index ebb8178d10..c106da61e3 100644 --- a/dubbo-demo/dubbo-demo-spring-boot/dubbo-demo-spring-boot-provider/pom.xml +++ b/dubbo-demo/dubbo-demo-spring-boot/dubbo-demo-spring-boot-provider/pom.xml @@ -102,6 +102,13 @@ org.springframework.boot spring-boot-starter-logging ${spring-boot.version} + + + + org.apache.logging.log4j + log4j-api + + diff --git a/dubbo-dependencies-bom/pom.xml b/dubbo-dependencies-bom/pom.xml index 24d4714ead..c3ffc1f834 100644 --- a/dubbo-dependencies-bom/pom.xml +++ b/dubbo-dependencies-bom/pom.xml @@ -139,7 +139,8 @@ 1.2 1.2.16 1.2.2 - 2.11.1 + + 2.17.0 2.6 0.10.0 @@ -168,7 +169,7 @@ 6.1.26 2.0 1.1.0 - 3.0.5-SNAPSHOT + 3.0.5 diff --git a/dubbo-dependencies/dubbo-dependencies-zookeeper/pom.xml b/dubbo-dependencies/dubbo-dependencies-zookeeper/pom.xml index 8e9acf0a0f..57485c3c4f 100644 --- a/dubbo-dependencies/dubbo-dependencies-zookeeper/pom.xml +++ b/dubbo-dependencies/dubbo-dependencies-zookeeper/pom.xml @@ -32,7 +32,7 @@ pom - 3.0.5-SNAPSHOT + 3.0.5 1.1.0 diff --git a/dubbo-plugin/dubbo-auth/src/test/java/org/apache/dubbo/auth/filter/ProviderAuthFilterTest.java b/dubbo-plugin/dubbo-auth/src/test/java/org/apache/dubbo/auth/filter/ProviderAuthFilterTest.java index aac62875b2..6e12f6fe56 100644 --- a/dubbo-plugin/dubbo-auth/src/test/java/org/apache/dubbo/auth/filter/ProviderAuthFilterTest.java +++ b/dubbo-plugin/dubbo-auth/src/test/java/org/apache/dubbo/auth/filter/ProviderAuthFilterTest.java @@ -24,7 +24,9 @@ import org.apache.dubbo.common.constants.CommonConstants; import org.apache.dubbo.rpc.Invocation; import org.apache.dubbo.rpc.Invoker; import org.apache.dubbo.rpc.Result; +import org.apache.dubbo.rpc.RpcInvocation; import org.apache.dubbo.rpc.model.ApplicationModel; + import org.junit.jupiter.api.Test; import static org.junit.jupiter.api.Assertions.assertNull; @@ -43,7 +45,7 @@ class ProviderAuthFilterTest { void testAuthDisabled() { URL url = mock(URL.class); Invoker invoker = mock(Invoker.class); - Invocation invocation = mock(Invocation.class); + Invocation invocation = mock(RpcInvocation.class); when(invoker.getUrl()).thenReturn(url); ProviderAuthFilter providerAuthFilter = new ProviderAuthFilter(ApplicationModel.defaultModel()); providerAuthFilter.invoke(invoker, invocation); @@ -58,7 +60,7 @@ class ProviderAuthFilterTest { .addParameter(CommonConstants.APPLICATION_KEY, "test") .addParameter(Constants.SERVICE_AUTH, true); Invoker invoker = mock(Invoker.class); - Invocation invocation = mock(Invocation.class); + Invocation invocation = mock(RpcInvocation.class); when(invoker.getUrl()).thenReturn(url); ProviderAuthFilter providerAuthFilter = new ProviderAuthFilter(ApplicationModel.defaultModel()); providerAuthFilter.invoke(invoker, invocation); @@ -74,7 +76,7 @@ class ProviderAuthFilterTest { .addParameter(CommonConstants.APPLICATION_KEY, "test") .addParameter(Constants.SERVICE_AUTH, true); Invoker invoker = mock(Invoker.class); - Invocation invocation = mock(Invocation.class); + Invocation invocation = mock(RpcInvocation.class); when(invocation.getAttachment(Constants.REQUEST_SIGNATURE_KEY)).thenReturn(null); when(invoker.getUrl()).thenReturn(url); @@ -92,7 +94,7 @@ class ProviderAuthFilterTest { .addParameter(CommonConstants.APPLICATION_KEY, "test") .addParameter(Constants.SERVICE_AUTH, true); Invoker invoker = mock(Invoker.class); - Invocation invocation = mock(Invocation.class); + Invocation invocation = mock(RpcInvocation.class); when(invocation.getAttachment(Constants.REQUEST_SIGNATURE_KEY)).thenReturn(null); when(invoker.getUrl()).thenReturn(url); @@ -107,7 +109,7 @@ class ProviderAuthFilterTest { .addParameter(CommonConstants.APPLICATION_KEY, "test-provider") .addParameter(Constants.SERVICE_AUTH, true); Invoker invoker = mock(Invoker.class); - Invocation invocation = mock(Invocation.class); + Invocation invocation = mock(RpcInvocation.class); when(invocation.getObjectAttachment(Constants.REQUEST_SIGNATURE_KEY)).thenReturn("dubbo"); when(invocation.getObjectAttachment(Constants.AK_KEY)).thenReturn("ak"); when(invocation.getObjectAttachment(CommonConstants.CONSUMER)).thenReturn("test-consumer"); @@ -135,7 +137,7 @@ class ProviderAuthFilterTest { .addParameter(Constants.SERVICE_AUTH, true); Invoker invoker = mock(Invoker.class); - Invocation invocation = mock(Invocation.class); + Invocation invocation = mock(RpcInvocation.class); when(invocation.getObjectAttachment(Constants.AK_KEY)).thenReturn("ak"); when(invocation.getObjectAttachment(CommonConstants.CONSUMER)).thenReturn("test-consumer"); when(invocation.getObjectAttachment(Constants.REQUEST_TIMESTAMP_KEY)).thenReturn(currentTimeMillis); @@ -168,7 +170,7 @@ class ProviderAuthFilterTest { .addParameter(CommonConstants.APPLICATION_KEY, "test-provider") .addParameter(Constants.SERVICE_AUTH, true); Invoker invoker = mock(Invoker.class); - Invocation invocation = mock(Invocation.class); + Invocation invocation = mock(RpcInvocation.class); when(invocation.getAttachment(Constants.AK_KEY)).thenReturn("ak"); when(invocation.getAttachment(CommonConstants.CONSUMER)).thenReturn("test-consumer"); when(invocation.getAttachment(Constants.REQUEST_TIMESTAMP_KEY)).thenReturn(String.valueOf(currentTimeMillis)); diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/InstanceAddressURL.java b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/InstanceAddressURL.java index 603726120a..3480f36359 100644 --- a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/InstanceAddressURL.java +++ b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/InstanceAddressURL.java @@ -25,7 +25,6 @@ import org.apache.dubbo.metadata.MetadataInfo; import org.apache.dubbo.rpc.RpcContext; import org.apache.dubbo.rpc.model.ApplicationModel; import org.apache.dubbo.rpc.model.FrameworkModel; -import org.apache.dubbo.rpc.model.ScopeModel; import org.apache.dubbo.rpc.model.ServiceModel; import java.util.HashMap; @@ -496,11 +495,6 @@ public class InstanceAddressURL extends URL { return this.instance.getMetadata(); } - @Override - public ScopeModel getScopeModel() { - return RpcContext.getServiceContext().getConsumerUrl().getScopeModel(); - } - @Override public FrameworkModel getOrDefaultFrameworkModel() { return instance.getOrDefaultApplicationModel().getFrameworkModel(); diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistryDirectory.java b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistryDirectory.java index 82c0f43227..47fd53b40c 100644 --- a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistryDirectory.java +++ b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistryDirectory.java @@ -198,17 +198,18 @@ public class ServiceDiscoveryRegistryDirectory extends DynamicDirectory { } private void refreshInvoker(List invokerUrls) { - Assert.notNull(invokerUrls, "invokerUrls should not be null, use empty url list to clear address."); + Assert.notNull(invokerUrls, "invokerUrls should not be null, use EMPTY url to clear current addresses."); this.originalUrls = invokerUrls; - if (invokerUrls.size() == 0) { - logger.info("Received empty url list..."); + if (invokerUrls.size() == 1 && EMPTY_PROTOCOL.equals(invokerUrls.get(0).getProtocol())) { + logger.warn("Received url with EMPTY protocol, will clear all available addresses."); this.forbidden = true; // Forbid to access routerChain.setInvokers(BitList.emptyList()); destroyAllInvokers(); // Close all invokers } else { this.forbidden = false; // Allow accessing if (CollectionUtils.isEmpty(invokerUrls)) { + logger.warn("Received empty url list, will ignore for protection purpose."); return; } diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/event/listener/ServiceInstancesChangedListener.java b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/event/listener/ServiceInstancesChangedListener.java index f7f5aaa3e9..983ef5724e 100644 --- a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/event/listener/ServiceInstancesChangedListener.java +++ b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/event/listener/ServiceInstancesChangedListener.java @@ -17,6 +17,7 @@ package org.apache.dubbo.registry.client.event.listener; import org.apache.dubbo.common.URL; +import org.apache.dubbo.common.URLBuilder; import org.apache.dubbo.common.logger.Logger; import org.apache.dubbo.common.logger.LoggerFactory; import org.apache.dubbo.common.threadpool.manager.ExecutorRepository; @@ -56,6 +57,8 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import static org.apache.dubbo.common.constants.CommonConstants.REMOTE_METADATA_STORAGE_TYPE; +import static org.apache.dubbo.common.constants.RegistryConstants.EMPTY_PROTOCOL; +import static org.apache.dubbo.common.constants.RegistryConstants.ENABLE_EMPTY_PROTECTION_KEY; import static org.apache.dubbo.metadata.RevisionResolver.EMPTY_REVISION; import static org.apache.dubbo.registry.client.metadata.ServiceInstanceMetadataUtils.getExportedServicesRevision; @@ -418,7 +421,7 @@ public class ServiceInstancesChangedListener { continue; } } - urls.add(i.toURL()); + urls.add(i.toURL().setScopeModel(i.getApplicationModel())); } } return urls; @@ -443,6 +446,14 @@ public class ServiceInstancesChangedListener { if (urls == null) { urls = Collections.emptyList(); } + boolean emptyProtectionEnabled = serviceDiscovery.getUrl().getParameter(ENABLE_EMPTY_PROTECTION_KEY, true); + if (CollectionUtils.isEmpty(urls) && !emptyProtectionEnabled) { + // notice that the service of this.url may not be the same as notify listener. + URL empty = URLBuilder.from(this.url) + .setProtocol(EMPTY_PROTOCOL) + .build(); + urls.add(empty); + } return urls; } diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/RegistryDirectory.java b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/RegistryDirectory.java index e53c900aa5..aa4fd38cea 100644 --- a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/RegistryDirectory.java +++ b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/RegistryDirectory.java @@ -214,6 +214,7 @@ public class RegistryDirectory extends DynamicDirectory { // use local reference to avoid NPE as this.cachedInvokerUrls will be set null by destroyAllInvokers(). Set localCachedInvokerUrls = this.cachedInvokerUrls; if (invokerUrls.isEmpty() && localCachedInvokerUrls != null) { + logger.warn("Service" + serviceKey + " received empty address list with no EMPTY protocol set, trigger empty protection."); invokerUrls.addAll(localCachedInvokerUrls); } else { localCachedInvokerUrls = new HashSet<>(); diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/support/CacheableFailbackRegistry.java b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/support/CacheableFailbackRegistry.java index ea7c56ca75..d4d94b8ea0 100644 --- a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/support/CacheableFailbackRegistry.java +++ b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/support/CacheableFailbackRegistry.java @@ -58,6 +58,7 @@ import static org.apache.dubbo.common.constants.CommonConstants.PATH_SEPARATOR; import static org.apache.dubbo.common.constants.CommonConstants.PROTOCOL_SEPARATOR_ENCODED; import static org.apache.dubbo.common.constants.RegistryConstants.CATEGORY_KEY; import static org.apache.dubbo.common.constants.RegistryConstants.EMPTY_PROTOCOL; +import static org.apache.dubbo.common.constants.RegistryConstants.ENABLE_EMPTY_PROTECTION_KEY; import static org.apache.dubbo.common.constants.RegistryConstants.PROVIDERS_CATEGORY; /** @@ -185,11 +186,16 @@ public abstract class CacheableFailbackRegistry extends FailbackRegistry { if (urls.isEmpty()) { int i = path.lastIndexOf(PATH_SEPARATOR); String category = i < 0 ? path : path.substring(i + 1); - URL empty = URLBuilder.from(consumer) + if (!PROVIDERS_CATEGORY.equals(category) || !getUrl().getParameter(ENABLE_EMPTY_PROTECTION_KEY, true)) { + if (PROVIDERS_CATEGORY.equals(category)) { + logger.warn("Service " + consumer.getServiceKey() + " received empty address list and empty protection is disabled, will clear current available addresses"); + } + URL empty = URLBuilder.from(consumer) .setProtocol(EMPTY_PROTOCOL) .addParameter(CATEGORY_KEY, category) .build(); - urls.add(empty); + urls.add(empty); + } } return urls; diff --git a/dubbo-registry/dubbo-registry-api/src/test/java/org/apache/dubbo/registry/CacheableFailbackRegistryTest.java b/dubbo-registry/dubbo-registry-api/src/test/java/org/apache/dubbo/registry/CacheableFailbackRegistryTest.java index 118e86fc10..0e52a74aa6 100644 --- a/dubbo-registry/dubbo-registry-api/src/test/java/org/apache/dubbo/registry/CacheableFailbackRegistryTest.java +++ b/dubbo-registry/dubbo-registry-api/src/test/java/org/apache/dubbo/registry/CacheableFailbackRegistryTest.java @@ -18,14 +18,20 @@ package org.apache.dubbo.registry; import org.apache.dubbo.common.URL; import org.apache.dubbo.common.URLStrParser; +import org.apache.dubbo.common.utils.CollectionUtils; import org.apache.dubbo.rpc.model.FrameworkModel; + import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import java.util.ArrayList; +import java.util.List; import java.util.concurrent.atomic.AtomicReference; +import static org.apache.dubbo.common.constants.RegistryConstants.EMPTY_PROTOCOL; +import static org.apache.dubbo.common.constants.RegistryConstants.ENABLE_EMPTY_PROTECTION_KEY; import static org.junit.jupiter.api.Assertions.assertEquals; public class CacheableFailbackRegistryTest { @@ -34,6 +40,9 @@ public class CacheableFailbackRegistryTest { static URL serviceUrl; static URL registryUrl; static String urlStr; + static String urlStr2; + static String urlStr3; + MockCacheableRegistryImpl registry; @BeforeAll @@ -49,6 +58,8 @@ public class CacheableFailbackRegistryTest { serviceUrl = URL.valueOf("dubbo://127.0.0.1/org.apache.dubbo.test.DemoService?category=providers"); registryUrl = URL.valueOf("http://1.2.3.4:9090/registry?check=false&file=N/A"); urlStr = "dubbo%3A%2F%2F172.19.4.113%3A20880%2Forg.apache.dubbo.demo.DemoService%3Fside%3Dprovider%26timeout%3D3000"; + urlStr2 = "dubbo%3A%2F%2F172.19.4.114%3A20880%2Forg.apache.dubbo.demo.DemoService%3Fside%3Dprovider%26timeout%3D3000"; + urlStr3 = "dubbo%3A%2F%2F172.19.4.115%3A20880%2Forg.apache.dubbo.demo.DemoService%3Fside%3Dprovider%26timeout%3D3000"; } @AfterEach @@ -181,4 +192,53 @@ public class CacheableFailbackRegistryTest { assertEquals(0, registry.getStringParam().size()); } + @Test + public void testEmptyProtection() { + final AtomicReference resCount = new AtomicReference<>(0); + final AtomicReference> currentUrls = new AtomicReference<>(); + final List EMPTY_LIST = new ArrayList<>(); + + registry = new MockCacheableRegistryImpl(registryUrl); + URL url = URLStrParser.parseEncodedStr(urlStr); + URL url2 = URLStrParser.parseEncodedStr(urlStr2); + URL url3 = URLStrParser.parseEncodedStr(urlStr3); + + NotifyListener listener = urls -> { + if (CollectionUtils.isEmpty(urls)) { + // do nothing + } else if (urls.size() == 1 && urls.get(0).getProtocol().equals(EMPTY_PROTOCOL)) { + resCount.set(0); + currentUrls.set(EMPTY_LIST); + } else { + resCount.set(urls.size()); + currentUrls.set(urls); + } + }; + + registry.addChildren(url); + registry.addChildren(url2); + registry.addChildren(url3); + + registry.subscribe(serviceUrl, listener); + assertEquals(3, resCount.get()); + registry.removeChildren(url); + assertEquals(2, resCount.get()); + registry.clearChildren(); + assertEquals(2, resCount.get()); + + URL emptyRegistryURL = registryUrl.addParameter(ENABLE_EMPTY_PROTECTION_KEY, false); + MockCacheableRegistryImpl emptyRegistry = new MockCacheableRegistryImpl(emptyRegistryURL); + + emptyRegistry.addChildren(url); + emptyRegistry.addChildren(url2); + + emptyRegistry.subscribe(serviceUrl, listener); + assertEquals(2, resCount.get()); + emptyRegistry.clearChildren(); + assertEquals(0, currentUrls.get().size()); + assertEquals(EMPTY_LIST, currentUrls.get()); + + } + + } diff --git a/dubbo-registry/dubbo-registry-api/src/test/java/org/apache/dubbo/registry/MockCacheableRegistryImpl.java b/dubbo-registry/dubbo-registry-api/src/test/java/org/apache/dubbo/registry/MockCacheableRegistryImpl.java index 1498c1ce0d..50fa724a97 100644 --- a/dubbo-registry/dubbo-registry-api/src/test/java/org/apache/dubbo/registry/MockCacheableRegistryImpl.java +++ b/dubbo-registry/dubbo-registry-api/src/test/java/org/apache/dubbo/registry/MockCacheableRegistryImpl.java @@ -34,6 +34,7 @@ import java.util.concurrent.Semaphore; public class MockCacheableRegistryImpl extends CacheableFailbackRegistry { private final List children = new ArrayList<>(); + NotifyListener listener; public MockCacheableRegistryImpl(URL url) { super(url); @@ -71,6 +72,7 @@ public class MockCacheableRegistryImpl extends CacheableFailbackRegistry { } } listener.notify(res); + this.listener = listener; } @Override @@ -87,12 +89,22 @@ public class MockCacheableRegistryImpl extends CacheableFailbackRegistry { children.add(URL.encode(url.toFullString())); } + public void removeChildren(URL url) { + children.remove(URL.encode(url.toFullString())); + if (listener != null) { + listener.notify(toUrlsWithEmpty(getUrl(), "providers", children)); + } + } + public List getChildren() { return children; } public void clearChildren() { children.clear(); + if (listener != null) { + listener.notify(toUrlsWithEmpty(getUrl(), "providers", children)); + } } public Map> getStringUrls() { diff --git a/dubbo-registry/dubbo-registry-api/src/test/java/org/apache/dubbo/registry/client/event/listener/ServiceInstancesChangedListenerTest.java b/dubbo-registry/dubbo-registry-api/src/test/java/org/apache/dubbo/registry/client/event/listener/ServiceInstancesChangedListenerTest.java index bfa36cd384..867face581 100644 --- a/dubbo-registry/dubbo-registry-api/src/test/java/org/apache/dubbo/registry/client/event/listener/ServiceInstancesChangedListenerTest.java +++ b/dubbo-registry/dubbo-registry-api/src/test/java/org/apache/dubbo/registry/client/event/listener/ServiceInstancesChangedListenerTest.java @@ -29,6 +29,7 @@ import org.apache.dubbo.registry.client.event.ServiceInstancesChangedEvent; import org.apache.dubbo.registry.client.metadata.MetadataUtils; import com.google.gson.Gson; +import org.apache.dubbo.rpc.model.ApplicationModel; import org.hamcrest.Matchers; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Assertions; @@ -99,6 +100,7 @@ public class ServiceInstancesChangedListenerTest { static String service3 = "org.apache.dubbo.demo.DemoService3"; static URL consumerURL = URL.valueOf("dubbo://127.0.0.1/org.apache.dubbo.demo.DemoService?registry_cluster=default"); + static URL registryURL = URL.valueOf("dubbo://127.0.0.1:2181/org.apache.dubbo.demo.RegistryService"); static MetadataInfo metadataInfo_111; static MetadataInfo metadataInfo_222; @@ -155,6 +157,7 @@ public class ServiceInstancesChangedListenerTest { Mockito.doThrow(IllegalStateException.class).when(metadataService).getMetadataInfo("444"); serviceDiscovery = Mockito.mock(ServiceDiscovery.class); + Mockito.doReturn(registryURL).when(serviceDiscovery).getUrl(); } @AfterEach @@ -328,7 +331,7 @@ public class ServiceInstancesChangedListenerTest { Assertions.assertEquals(0, revisionToMetadata_app2.size()); assertTrue(isEmpty(listener.getAddresses(service1 + ":dubbo", consumerURL))); - assertTrue(isEmpty(listener.getAddresses(service2+ ":dubbo", consumerURL))); + assertTrue(isEmpty(listener.getAddresses(service2 + ":dubbo", consumerURL))); assertTrue(isEmpty(listener.getAddresses(service3 + ":dubbo", consumerURL))); } } @@ -511,7 +514,7 @@ public class ServiceInstancesChangedListenerTest { List instances = new ArrayList<>(); for (Object obj : rawURls) { - String rawURL = (String)obj; + String rawURL = (String) obj; DefaultServiceInstance instance = new DefaultServiceInstance(); final URL dubboUrl = URL.valueOf(rawURL); instance.setRawAddress(rawURL); @@ -520,6 +523,7 @@ public class ServiceInstancesChangedListenerTest { instance.setHealthy(true); instance.setPort(dubboUrl.getPort()); instance.setRegistryCluster("default"); + instance.setApplicationModel(ApplicationModel.defaultModel()); Map metadata = new HashMap<>(); if (StringUtils.isNotEmpty(dubboUrl.getParameter(REVISION_KEY))) { diff --git a/dubbo-registry/dubbo-registry-multiple/src/main/java/org/apache/dubbo/registry/multiple/MultipleRegistry.java b/dubbo-registry/dubbo-registry-multiple/src/main/java/org/apache/dubbo/registry/multiple/MultipleRegistry.java index e3e226a9c3..0e83e7034f 100644 --- a/dubbo-registry/dubbo-registry-multiple/src/main/java/org/apache/dubbo/registry/multiple/MultipleRegistry.java +++ b/dubbo-registry/dubbo-registry-multiple/src/main/java/org/apache/dubbo/registry/multiple/MultipleRegistry.java @@ -93,7 +93,7 @@ public class MultipleRegistry extends AbstractRegistry { serviceRegistries.put(tmpUrl, registryMap.get(tmpUrl)); continue; } - final URL registryUrl = URL.valueOf(tmpUrl).addParameterIfAbsent(CHECK_KEY, url.getParameter(CHECK_KEY, "true")); + final URL registryUrl = URL.valueOf(tmpUrl).addParametersIfAbsent(url.getParameters()).addParameterIfAbsent(CHECK_KEY, url.getParameter(CHECK_KEY, "true")); Registry registry = registryFactory.getRegistry(registryUrl); registryMap.put(tmpUrl, registry); serviceRegistries.put(tmpUrl, registry); diff --git a/dubbo-registry/dubbo-registry-multiple/src/test/java/org/apache/dubbo/registry/multiple/MultipleRegistry2S2RTest.java b/dubbo-registry/dubbo-registry-multiple/src/test/java/org/apache/dubbo/registry/multiple/MultipleRegistry2S2RTest.java index d60af42371..f16502e2e8 100644 --- a/dubbo-registry/dubbo-registry-multiple/src/test/java/org/apache/dubbo/registry/multiple/MultipleRegistry2S2RTest.java +++ b/dubbo-registry/dubbo-registry-multiple/src/test/java/org/apache/dubbo/registry/multiple/MultipleRegistry2S2RTest.java @@ -22,9 +22,11 @@ import org.apache.dubbo.registry.Registry; import org.apache.dubbo.registry.zookeeper.ZookeeperRegistry; import org.apache.dubbo.remoting.zookeeper.ZookeeperClient; import org.apache.dubbo.remoting.zookeeper.curator.CuratorZookeeperClient; + import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; + import java.util.ArrayList; import java.util.List; @@ -52,7 +54,7 @@ public class MultipleRegistry2S2RTest { zookeeperConnectionAddress1 = System.getProperty("zookeeper.connection.address.1"); zookeeperConnectionAddress2 = System.getProperty("zookeeper.connection.address.2"); - URL url = URL.valueOf("multiple://127.0.0.1?application=vic&" + + URL url = URL.valueOf("multiple://127.0.0.1?application=vic&enable-empty-protection=false&" + MultipleRegistry.REGISTRY_FOR_SERVICE + "=" + zookeeperConnectionAddress1 + "," + zookeeperConnectionAddress2 + "&" + MultipleRegistry.REGISTRY_FOR_REFERENCE + "=" + zookeeperConnectionAddress1 + "," + zookeeperConnectionAddress2); multipleRegistry = (MultipleRegistry) new MultipleRegistryFactory().createRegistry(url); diff --git a/dubbo-registry/dubbo-registry-nacos/src/main/java/org/apache/dubbo/registry/nacos/NacosRegistry.java b/dubbo-registry/dubbo-registry-nacos/src/main/java/org/apache/dubbo/registry/nacos/NacosRegistry.java index 7a0605ac18..8405caf599 100644 --- a/dubbo-registry/dubbo-registry-nacos/src/main/java/org/apache/dubbo/registry/nacos/NacosRegistry.java +++ b/dubbo-registry/dubbo-registry-nacos/src/main/java/org/apache/dubbo/registry/nacos/NacosRegistry.java @@ -49,13 +49,13 @@ import java.util.LinkedHashSet; import java.util.LinkedList; import java.util.List; import java.util.Map; -import java.util.Set; import java.util.Objects; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; -import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.ConcurrentHashMap; import java.util.stream.Collectors; import static org.apache.dubbo.common.constants.CommonConstants.ANY_VALUE; @@ -70,6 +70,7 @@ import static org.apache.dubbo.common.constants.RegistryConstants.CONFIGURATORS_ import static org.apache.dubbo.common.constants.RegistryConstants.CONSUMERS_CATEGORY; import static org.apache.dubbo.common.constants.RegistryConstants.DEFAULT_CATEGORY; import static org.apache.dubbo.common.constants.RegistryConstants.EMPTY_PROTOCOL; +import static org.apache.dubbo.common.constants.RegistryConstants.ENABLE_EMPTY_PROTECTION_KEY; import static org.apache.dubbo.common.constants.RegistryConstants.PROVIDERS_CATEGORY; import static org.apache.dubbo.common.constants.RegistryConstants.ROUTERS_CATEGORY; import static org.apache.dubbo.registry.Constants.ADMIN_PROTOCOL; @@ -497,7 +498,9 @@ public class NacosRegistry extends FailbackRegistry { private List toUrlWithEmpty(URL consumerURL, Collection instances) { List urls = buildURLs(consumerURL, instances); - if (urls.size() == 0) { + // Nacos does not support configurators and routers from registry, so all notifications are of providers type. + if (urls.size() == 0 && !getUrl().getParameter(ENABLE_EMPTY_PROTECTION_KEY, true)) { + logger.warn("Received empty url address list and empty protection is disabled, will clear current available addresses"); URL empty = URLBuilder.from(consumerURL) .setProtocol(EMPTY_PROTOCOL) .addParameter(CATEGORY_KEY, DEFAULT_CATEGORY) diff --git a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/transport/AbstractEndpoint.java b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/transport/AbstractEndpoint.java index cb442d61dd..fbba37839b 100644 --- a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/transport/AbstractEndpoint.java +++ b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/transport/AbstractEndpoint.java @@ -20,6 +20,7 @@ import org.apache.dubbo.common.Resetable; import org.apache.dubbo.common.URL; import org.apache.dubbo.common.logger.Logger; import org.apache.dubbo.common.logger.LoggerFactory; +import org.apache.dubbo.common.utils.StringUtils; import org.apache.dubbo.remoting.ChannelHandler; import org.apache.dubbo.remoting.Codec; import org.apache.dubbo.remoting.Codec2; @@ -47,13 +48,17 @@ public abstract class AbstractEndpoint extends AbstractPeer implements Resetable } protected static Codec2 getChannelCodec(URL url) { - String codecName = url.getProtocol(); // codec extension name must stay the same with protocol name + String codecName = url.getParameter(Constants.CODEC_KEY); + if (StringUtils.isEmpty(codecName)) { + // codec extension name must stay the same with protocol name + codecName = url.getProtocol(); + } FrameworkModel frameworkModel = getFrameworkModel(url.getScopeModel()); if (frameworkModel.getExtensionLoader(Codec2.class).hasExtension(codecName)) { return frameworkModel.getExtensionLoader(Codec2.class).getExtension(codecName); } else { return new CodecAdapter(frameworkModel.getExtensionLoader(Codec.class) - .getExtension(codecName)); + .getExtension(codecName)); } } @@ -61,7 +66,7 @@ public abstract class AbstractEndpoint extends AbstractPeer implements Resetable public void reset(URL url) { if (isClosed()) { throw new IllegalStateException("Failed to reset parameters " - + url + ", cause: Channel closed. channel: " + getLocalAddress()); + + url + ", cause: Channel closed. channel: " + getLocalAddress()); } try { diff --git a/dubbo-remoting/dubbo-remoting-netty/src/test/java/org/apache/dubbo/remoting/exchange/support/header/HeartbeatHandlerTest.java b/dubbo-remoting/dubbo-remoting-netty/src/test/java/org/apache/dubbo/remoting/exchange/support/header/HeartbeatHandlerTest.java index 9836dff2f3..3fbac2286a 100644 --- a/dubbo-remoting/dubbo-remoting-netty/src/test/java/org/apache/dubbo/remoting/exchange/support/header/HeartbeatHandlerTest.java +++ b/dubbo-remoting/dubbo-remoting-netty/src/test/java/org/apache/dubbo/remoting/exchange/support/header/HeartbeatHandlerTest.java @@ -67,9 +67,9 @@ public class HeartbeatHandlerTest { public void testServerHeartbeat() throws Exception { FakeChannelHandlers.resetChannelHandlers(); URL serverURL = URL.valueOf("telnet://localhost:" + NetUtils.getAvailablePort(56780)) - .addParameter(Constants.EXCHANGER_KEY, HeaderExchanger.NAME) - .addParameter(Constants.TRANSPORTER_KEY, "netty3") - .addParameter(Constants.HEARTBEAT_KEY, 1000); + .addParameter(Constants.EXCHANGER_KEY, HeaderExchanger.NAME) + .addParameter(Constants.TRANSPORTER_KEY, "netty3") + .addParameter(Constants.HEARTBEAT_KEY, 1000); CountDownLatch connect = new CountDownLatch(1); CountDownLatch disconnect = new CountDownLatch(1); TestHeartbeatHandler handler = new TestHeartbeatHandler(connect, disconnect); @@ -82,6 +82,7 @@ public class HeartbeatHandlerTest { // Let the client not reply to the heartbeat, and turn off automatic reconnect to simulate the client dropped. serverURL = serverURL.addParameter(Constants.HEARTBEAT_KEY, 600 * 1000); serverURL = serverURL.addParameter(Constants.RECONNECT_KEY, false); + serverURL = serverURL.addParameter(Constants.CODEC_KEY, "telnet"); client = Exchangers.connect(serverURL); disconnect.await(); @@ -92,9 +93,10 @@ public class HeartbeatHandlerTest { @Test public void testHeartbeat() throws Exception { URL serverURL = URL.valueOf("telnet://localhost:" + NetUtils.getAvailablePort(56785)) - .addParameter(Constants.EXCHANGER_KEY, HeaderExchanger.NAME) - .addParameter(Constants.TRANSPORTER_KEY, "netty3") - .addParameter(Constants.HEARTBEAT_KEY, 1000); + .addParameter(Constants.EXCHANGER_KEY, HeaderExchanger.NAME) + .addParameter(Constants.TRANSPORTER_KEY, "netty3") + .addParameter(Constants.HEARTBEAT_KEY, 1000) + .addParameter(Constants.CODEC_KEY, "telnet"); CountDownLatch connect = new CountDownLatch(1); CountDownLatch disconnect = new CountDownLatch(1); TestHeartbeatHandler handler = new TestHeartbeatHandler(connect, disconnect); @@ -113,8 +115,9 @@ public class HeartbeatHandlerTest { public void testClientHeartbeat() throws Exception { FakeChannelHandlers.setTestingChannelHandlers(); URL serverURL = URL.valueOf("telnet://localhost:" + NetUtils.getAvailablePort(56790)) - .addParameter(Constants.EXCHANGER_KEY, HeaderExchanger.NAME) - .addParameter(Constants.TRANSPORTER_KEY, "netty3"); + .addParameter(Constants.EXCHANGER_KEY, HeaderExchanger.NAME) + .addParameter(Constants.TRANSPORTER_KEY, "netty3") + .addParameter(Constants.CODEC_KEY, "telnet"); CountDownLatch connect = new CountDownLatch(1); CountDownLatch disconnect = new CountDownLatch(1); TestHeartbeatHandler handler = new TestHeartbeatHandler(connect, disconnect); diff --git a/dubbo-remoting/dubbo-remoting-netty/src/test/java/org/apache/dubbo/remoting/transport/netty/ClientReconnectTest.java b/dubbo-remoting/dubbo-remoting-netty/src/test/java/org/apache/dubbo/remoting/transport/netty/ClientReconnectTest.java index 167647a4b4..edc65c901c 100644 --- a/dubbo-remoting/dubbo-remoting-netty/src/test/java/org/apache/dubbo/remoting/transport/netty/ClientReconnectTest.java +++ b/dubbo-remoting/dubbo-remoting-netty/src/test/java/org/apache/dubbo/remoting/transport/netty/ClientReconnectTest.java @@ -70,7 +70,7 @@ public class ClientReconnectTest { public Client startClient(int port, int heartbeat) throws RemotingException { - final String url = "exchange://127.0.0.1:" + port + "/client.reconnect.test?check=false&client=netty3&" + + final String url = "exchange://127.0.0.1:" + port + "/client.reconnect.test?check=false&codec=exchange&client=netty3&" + Constants.HEARTBEAT_KEY + "=" + heartbeat; return Exchangers.connect(url); } diff --git a/dubbo-remoting/dubbo-remoting-netty/src/test/java/org/apache/dubbo/remoting/transport/netty/NettyClientTest.java b/dubbo-remoting/dubbo-remoting-netty/src/test/java/org/apache/dubbo/remoting/transport/netty/NettyClientTest.java index 6ac5374a8a..076ad34ece 100644 --- a/dubbo-remoting/dubbo-remoting-netty/src/test/java/org/apache/dubbo/remoting/transport/netty/NettyClientTest.java +++ b/dubbo-remoting/dubbo-remoting-netty/src/test/java/org/apache/dubbo/remoting/transport/netty/NettyClientTest.java @@ -40,7 +40,7 @@ public class NettyClientTest { @BeforeAll public static void setUp() throws Exception { - server = Exchangers.bind(URL.valueOf("exchange://localhost:" + port + "?server=netty3"), new TelnetServerHandler()); + server = Exchangers.bind(URL.valueOf("exchange://localhost:" + port + "?server=netty3&codec=exchange"), new TelnetServerHandler()); } @AfterAll @@ -53,7 +53,7 @@ public class NettyClientTest { } public static void main(String[] args) throws RemotingException, InterruptedException { - ExchangeChannel client = Exchangers.connect(URL.valueOf("exchange://10.20.153.10:20880?client=netty3&heartbeat=1000")); + ExchangeChannel client = Exchangers.connect(URL.valueOf("exchange://10.20.153.10:20880?client=netty3&heartbeat=1000&codec=exchange")); Thread.sleep(60 * 1000 * 50); } @@ -61,7 +61,7 @@ public class NettyClientTest { public void testClientClose() throws Exception { List clients = new ArrayList(100); for (int i = 0; i < 100; i++) { - ExchangeChannel client = Exchangers.connect(URL.valueOf("exchange://localhost:" + port + "?client=netty3")); + ExchangeChannel client = Exchangers.connect(URL.valueOf("exchange://localhost:" + port + "?client=netty3&codec=exchange")); Thread.sleep(5); clients.add(client); } @@ -74,7 +74,7 @@ public class NettyClientTest { @Test public void testServerClose() throws Exception { for (int i = 0; i < 100; i++) { - RemotingServer aServer = Exchangers.bind(URL.valueOf("exchange://localhost:" + NetUtils.getAvailablePort(6000) + "?server=netty3"), new TelnetServerHandler()); + RemotingServer aServer = Exchangers.bind(URL.valueOf("exchange://localhost:" + NetUtils.getAvailablePort(6000) + "?server=netty3&codec=exchange"), new TelnetServerHandler()); aServer.close(); } } diff --git a/dubbo-remoting/dubbo-remoting-netty/src/test/java/org/apache/dubbo/remoting/transport/netty/NettyClientToServerTest.java b/dubbo-remoting/dubbo-remoting-netty/src/test/java/org/apache/dubbo/remoting/transport/netty/NettyClientToServerTest.java index 4bc62df0b0..51d8bd4df7 100644 --- a/dubbo-remoting/dubbo-remoting-netty/src/test/java/org/apache/dubbo/remoting/transport/netty/NettyClientToServerTest.java +++ b/dubbo-remoting/dubbo-remoting-netty/src/test/java/org/apache/dubbo/remoting/transport/netty/NettyClientToServerTest.java @@ -31,14 +31,14 @@ public class NettyClientToServerTest extends ClientToServerTest { protected ExchangeServer newServer(int port, Replier receiver) throws RemotingException { // add heartbeat cycle to avoid unstable ut. - URL url = URL.valueOf("exchange://localhost:" + port + "?server=netty3"); + URL url = URL.valueOf("exchange://localhost:" + port + "?server=netty3&codec=exchange"); url = url.addParameter(Constants.HEARTBEAT_KEY, 600 * 1000); return Exchangers.bind(url, receiver); } protected ExchangeChannel newClient(int port) throws RemotingException { // add heartbeat cycle to avoid unstable ut. - URL url = URL.valueOf("exchange://localhost:" + port + "?client=netty3&timeout=3000"); + URL url = URL.valueOf("exchange://localhost:" + port + "?client=netty3&timeout=3000&codec=exchange"); url = url.addParameter(Constants.HEARTBEAT_KEY, 600 * 1000); return Exchangers.connect(url); } diff --git a/dubbo-remoting/dubbo-remoting-netty/src/test/java/org/apache/dubbo/remoting/transport/netty/NettyStringTest.java b/dubbo-remoting/dubbo-remoting-netty/src/test/java/org/apache/dubbo/remoting/transport/netty/NettyStringTest.java index d08d78153d..6781897472 100644 --- a/dubbo-remoting/dubbo-remoting-netty/src/test/java/org/apache/dubbo/remoting/transport/netty/NettyStringTest.java +++ b/dubbo-remoting/dubbo-remoting-netty/src/test/java/org/apache/dubbo/remoting/transport/netty/NettyStringTest.java @@ -40,8 +40,8 @@ public class NettyStringTest { //int port = 10001; int port = NetUtils.getAvailablePort(); System.out.println(port); - server = Exchangers.bind(URL.valueOf("telnet://0.0.0.0:" + port + "?server=netty3"), new TelnetServerHandler()); - client = Exchangers.connect(URL.valueOf("telnet://127.0.0.1:" + port + "?client=netty3"), new TelnetClientHandler()); + server = Exchangers.bind(URL.valueOf("telnet://0.0.0.0:" + port + "?server=netty3&codec=telnet"), new TelnetServerHandler()); + client = Exchangers.connect(URL.valueOf("telnet://127.0.0.1:" + port + "?client=netty3&codec=telnet"), new TelnetClientHandler()); } @AfterAll diff --git a/dubbo-remoting/dubbo-remoting-netty/src/test/java/org/apache/dubbo/remoting/transport/netty/ThreadNameTest.java b/dubbo-remoting/dubbo-remoting-netty/src/test/java/org/apache/dubbo/remoting/transport/netty/ThreadNameTest.java index abb02a8734..26c5edc5d5 100644 --- a/dubbo-remoting/dubbo-remoting-netty/src/test/java/org/apache/dubbo/remoting/transport/netty/ThreadNameTest.java +++ b/dubbo-remoting/dubbo-remoting-netty/src/test/java/org/apache/dubbo/remoting/transport/netty/ThreadNameTest.java @@ -51,10 +51,10 @@ public class ThreadNameTest { @BeforeEach public void before() throws Exception { int port = NetUtils.getAvailablePort(20880 + new Random().nextInt(10000)); - serverURL = URL.valueOf("telnet://localhost?side=provider") + serverURL = URL.valueOf("telnet://localhost?side=provider&codec=telnet") .setPort(port) .setScopeModel(ApplicationModel.defaultModel()); - clientURL = URL.valueOf("telnet://localhost?side=consumer") + clientURL = URL.valueOf("telnet://localhost?side=consumer&codec=telnet") .setPort(port) .setScopeModel(ApplicationModel.defaultModel()); diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/AsyncContext.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/AsyncContext.java index 1f51a54d4f..0635d290c2 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/AsyncContext.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/AsyncContext.java @@ -74,4 +74,50 @@ public interface AsyncContext { * */ void signalContextSwitch(); + + /** + * Reset Context is not necessary. Only reset context after result was write back if it is necessary. + * + * + * public class AsyncServiceImpl implements AsyncService { + * public String sayHello(String name) { + * final AsyncContext asyncContext = RpcContext.startAsync(); + * new Thread(() -> { + *

+ * // the right place to use this method + * asyncContext.signalContextSwitch(); + *

+ * try { + * Thread.sleep(500); + * } catch (InterruptedException e) { + * e.printStackTrace(); + * } + * asyncContext.write("Hello " + name + ", response from provider."); + * // only reset after asyncContext.write() + * asyncContext.resetContext(); + * }).start(); + * return null; + * } + * } + * + * + * + * public class AsyncServiceImpl implements AsyncService { + * public CompletableFuture sayHello(String name) { + * CompletableFuture future = new CompletableFuture(); + * final AsyncContext asyncContext = RpcContext.startAsync(); + * new Thread(() -> { + * // the right place to use this method + * asyncContext.signalContextSwitch(); + * // some operations... + * future.complete(); + * // only reset after future.complete() + * asyncContext.resetContext(); + * }).start(); + * return future; + * } + * } + * + */ + void resetContext(); } diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/AsyncContextImpl.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/AsyncContextImpl.java index 1331b21e0d..4ff3368aa0 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/AsyncContextImpl.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/AsyncContextImpl.java @@ -26,12 +26,13 @@ public class AsyncContextImpl implements AsyncContext { private CompletableFuture future; - private RpcContextAttachment storedContext; - private RpcContextAttachment storedServerContext; + private final RpcContext.RestoreContext restoreContext; + private final ClassLoader restoreClassLoader; + private ClassLoader stagedClassLoader; public AsyncContextImpl() { - this.storedContext = RpcContext.getClientAttachment(); - this.storedServerContext = RpcContext.getServerContext(); + restoreContext = RpcContext.storeContext(); + restoreClassLoader = Thread.currentThread().getContextClassLoader(); } @Override @@ -67,9 +68,19 @@ public class AsyncContextImpl implements AsyncContext { @Override public void signalContextSwitch() { - RpcContext.restoreContext(storedContext); - RpcContext.restoreServerContext(storedServerContext); - // Restore any other contexts in here if necessary. + RpcContext.restoreContext(restoreContext); + if (restoreClassLoader != null) { + stagedClassLoader = Thread.currentThread().getContextClassLoader(); + Thread.currentThread().setContextClassLoader(restoreClassLoader); + } + } + + @Override + public void resetContext() { + RpcContext.removeContext(); + if (stagedClassLoader != null) { + Thread.currentThread().setContextClassLoader(restoreClassLoader); + } } public CompletableFuture getInternalFuture() { diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/AsyncRpcResult.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/AsyncRpcResult.java index 1accccd5bc..377432a71c 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/AsyncRpcResult.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/AsyncRpcResult.java @@ -20,6 +20,7 @@ import org.apache.dubbo.common.logger.Logger; import org.apache.dubbo.common.logger.LoggerFactory; import org.apache.dubbo.common.threadpool.ThreadlessExecutor; import org.apache.dubbo.rpc.model.ConsumerMethodModel; +import org.apache.dubbo.rpc.protocol.dubbo.FutureAdapter; import java.util.Map; import java.util.concurrent.CompletableFuture; @@ -30,6 +31,7 @@ import java.util.concurrent.TimeoutException; import java.util.function.BiConsumer; import java.util.function.Function; +import static org.apache.dubbo.common.constants.CommonConstants.PROVIDER_ASYNC_KEY; import static org.apache.dubbo.common.utils.ReflectUtils.defaultReturn; /** @@ -51,21 +53,27 @@ public class AsyncRpcResult implements Result { /** * RpcContext may already have been changed when callback happens, it happens when the same thread is used to execute another RPC call. - * So we should keep the reference of current RpcContext instance and restore it before callback being executed. + * So we should keep the copy of current RpcContext instance and restore it before callback being executed. */ - private RpcContextAttachment storedContext; - private RpcContextAttachment storedServerContext; + private RpcContext.RestoreContext storedContext; + private Executor executor; private Invocation invocation; + private final boolean async; private CompletableFuture responseFuture; public AsyncRpcResult(CompletableFuture future, Invocation invocation) { this.responseFuture = future; this.invocation = invocation; - this.storedContext = RpcContext.getClientAttachment(); - this.storedServerContext = RpcContext.getServerContext(); + RpcInvocation rpcInvocation = (RpcInvocation) invocation; + if ((rpcInvocation.get(PROVIDER_ASYNC_KEY) != null || InvokeMode.SYNC != rpcInvocation.getInvokeMode()) && !future.isDone()) { + async = true; + this.storedContext = RpcContext.clearAndStoreContext(); + } else { + async = false; + } } /** @@ -193,10 +201,15 @@ public class AsyncRpcResult implements Result { public Result whenCompleteWithContext(BiConsumer fn) { this.responseFuture = this.responseFuture.whenComplete((v, t) -> { - beforeContext.accept(v, t); + if (async) { + RpcContext.restoreContext(storedContext); + } fn.accept(v, t); - afterContext.accept(v, t); }); + + // Necessary! update future in context, see https://github.com/apache/dubbo/issues/9461 + RpcContext.getServiceContext().setFuture(new FutureAdapter<>(this.responseFuture)); + return this; } @@ -280,24 +293,6 @@ public class AsyncRpcResult implements Result { this.executor = executor; } - /** - * tmp context to use when the thread switch to Dubbo thread. - */ - private RpcContextAttachment tmpContext; - - private RpcContextAttachment tmpServerContext; - private BiConsumer beforeContext = (appResponse, t) -> { - tmpContext = RpcContext.getClientAttachment(); - tmpServerContext = RpcContext.getServerContext(); - RpcContext.restoreContext(storedContext); - RpcContext.restoreServerContext(storedServerContext); - }; - - private BiConsumer afterContext = (appResponse, t) -> { - RpcContext.restoreContext(tmpContext); - RpcContext.restoreServerContext(tmpServerContext); - }; - /** * Some utility methods used to quickly generate default AsyncRpcResult instance. */ diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/BaseFilter.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/BaseFilter.java index 594954f5fc..02c643abf5 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/BaseFilter.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/BaseFilter.java @@ -22,10 +22,33 @@ public interface BaseFilter { */ Result invoke(Invoker invoker, Invocation invocation) throws RpcException; + /** + * This callback listener applies to both synchronous and asynchronous calls, please put logics that need to be executed + * on return of rpc result in onResponse or onError respectively based on it is normal return or exception return. + *

+ * There's something that needs to pay attention on legacy synchronous style filer refactor, the thing is, try to move logics + * previously defined in the 'finally block' to both onResponse and onError. + */ interface Listener { + /** + * This method will only be called on successful remote rpc execution, that means, the service in on remote received + * the request and the result (normal or exceptional) returned successfully. + * + * @param appResponse, the rpc call result, it can represent both normal result and exceptional result + * @param invoker, context + * @param invocation, context + */ void onResponse(Result appResponse, Invoker invoker, Invocation invocation); + /** + * This method will be called on detection of framework exceptions, for example, TimeoutException, NetworkException + * Exception raised in Filters, etc. + * + * @param t, framework exception + * @param invoker, context + * @param invocation, context + */ void onError(Throwable t, Invoker invoker, Invocation invocation); } } diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/ListenableFilter.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/ListenableFilter.java index 160ddcfe1b..b590bfbb35 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/ListenableFilter.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/ListenableFilter.java @@ -23,9 +23,10 @@ import java.util.concurrent.ConcurrentMap; * It's recommended to implement Filter.Listener directly for callback registration, check the default implementation, * see {@link org.apache.dubbo.rpc.filter.ExceptionFilter}, for example. *

- * If you do not want to share Listener instance between RPC calls. You can use ListenableFilter + * If you do not want to share Listener instance between RPC calls. ListenableFilter can be used * to keep a 'one Listener each RPC call' model. */ +@Deprecated public abstract class ListenableFilter implements Filter { protected Listener listener = null; diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcContext.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcContext.java index 13f108bc19..69e986eac5 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcContext.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcContext.java @@ -122,10 +122,6 @@ public class RpcContext { return SERVER_LOCAL.get(); } - public static void restoreServerContext(RpcContextAttachment oldServerContext) { - SERVER_LOCAL.set(oldServerContext); - } - /** * remove server side context. * @@ -198,10 +194,6 @@ public class RpcContext { this.remove = remove; } - public static void restoreContext(RpcContextAttachment oldContext) { - CLIENT_ATTACHMENT.set(oldContext); - } - /** * remove context. * @@ -810,4 +802,60 @@ public class RpcContext { public static void setRpcContext(URL url) { RpcServiceContext.setRpcContext(url); } + + protected static RestoreContext clearAndStoreContext() { + RestoreContext restoreContext = new RestoreContext(); + RpcContext.removeContext(); + return restoreContext; + } + + protected static RestoreContext storeContext() { + return new RestoreContext(); + } + + protected static void restoreContext(RestoreContext restoreContext) { + if (restoreContext != null) { + restoreContext.restore(); + } + } + + /** + * Used to temporarily store and restore all kinds of contexts of current thread. + */ + public static class RestoreContext { + private final RpcServiceContext serviceContext; + private final RpcContextAttachment clientAttachment; + private final RpcContextAttachment serverAttachment; + private final RpcContextAttachment serverLocal; + + public RestoreContext() { + serviceContext = getServiceContext().copyOf(false); + clientAttachment = getClientAttachment().copyOf(false); + serverAttachment = getServerAttachment().copyOf(false); + serverLocal = getServerContext().copyOf(false); + } + + public void restore() { + if (serviceContext != null) { + SERVICE_CONTEXT.set(serviceContext); + } else { + removeServiceContext(); + } + if (clientAttachment != null) { + CLIENT_ATTACHMENT.set(clientAttachment); + } else { + removeClientAttachment(); + } + if (serverAttachment != null) { + SERVER_ATTACHMENT.set(serverAttachment); + } else { + removeServerAttachment(); + } + if (serverLocal != null) { + SERVER_LOCAL.set(serverLocal); + } else { + removeServerContext(); + } + } + } } diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcContextAttachment.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcContextAttachment.java index ddbd9813a7..aec69433be 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcContextAttachment.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcContextAttachment.java @@ -201,4 +201,28 @@ public class RpcContextAttachment extends RpcContext{ return getAttachment(key); } + /** + * Also see {@link RpcServiceContext#copyOf(boolean)} + * + * @return a copy of RpcContextAttachment with deep copied attachments + */ + public RpcContextAttachment copyOf(boolean needCopy) { + if (!isValid()) { + return null; + } + + if (needCopy) { + RpcContextAttachment copy = new RpcContextAttachment(); + if (CollectionUtils.isNotEmptyMap(attachments)) { + copy.attachments.putAll(this.attachments); + } + return copy; + } else { + return this; + } + } + + private boolean isValid() { + return CollectionUtils.isNotEmptyMap(attachments); + } } diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcServiceContext.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcServiceContext.java index ebec31f7d1..e59ae9b313 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcServiceContext.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcServiceContext.java @@ -41,6 +41,9 @@ public class RpcServiceContext extends RpcContext { protected RpcServiceContext() { } + // RPC service context updated before each service call. + private URL consumerUrl; + private List urls; private URL url; @@ -584,9 +587,6 @@ public class RpcServiceContext extends RpcContext { return asyncContext; } - // RPC service context updated before each service call. - private URL consumerUrl; - @Override public String getGroup() { if (consumerUrl == null) { @@ -649,4 +649,26 @@ public class RpcServiceContext extends RpcContext { RpcServiceContext rpcContext = RpcContext.getServiceContext(); rpcContext.setConsumerUrl(url); } + + /** + * Only part of the properties are copied, the others are either not used currently or can be got from invocation. + * Also see {@link RpcContextAttachment#copyOf(boolean)} + * + * @param needCopy + * @return a shallow copy of RpcServiceContext + */ + public RpcServiceContext copyOf(boolean needCopy) { + if (needCopy) { + RpcServiceContext copy = new RpcServiceContext(); + copy.consumerUrl = this.consumerUrl; + copy.localAddress = this.localAddress; + copy.remoteAddress = this.remoteAddress; + copy.invocation = this.invocation; + copy.asyncContext = this.asyncContext; + return copy; + } else { + return this; + } + } + } diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ClassLoaderCallbackFilter.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ClassLoaderCallbackFilter.java new file mode 100644 index 0000000000..a9b1269447 --- /dev/null +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ClassLoaderCallbackFilter.java @@ -0,0 +1,57 @@ +/* + * 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.filter; + +import org.apache.dubbo.common.constants.CommonConstants; +import org.apache.dubbo.common.extension.Activate; +import org.apache.dubbo.rpc.BaseFilter; +import org.apache.dubbo.rpc.Filter; +import org.apache.dubbo.rpc.Invocation; +import org.apache.dubbo.rpc.Invoker; +import org.apache.dubbo.rpc.Result; +import org.apache.dubbo.rpc.RpcException; + +import static org.apache.dubbo.common.constants.CommonConstants.WORKING_CLASSLOADER_KEY; + +/** + * Switch thread context class loader on filter callback. + */ +@Activate(group = CommonConstants.PROVIDER, order = Integer.MAX_VALUE) +public class ClassLoaderCallbackFilter implements Filter, BaseFilter.Listener { + + @Override + public Result invoke(Invoker invoker, Invocation invocation) throws RpcException { + return invoker.invoke(invocation); + } + + @Override + public void onResponse(Result appResponse, Invoker invoker, Invocation invocation) { + setClassLoader(invoker, invocation); + } + + @Override + public void onError(Throwable t, Invoker invoker, Invocation invocation) { + setClassLoader(invoker, invocation); + } + + private void setClassLoader(Invoker invoker, Invocation invocation) { + ClassLoader workingClassLoader = (ClassLoader) invocation.get(WORKING_CLASSLOADER_KEY); + if (workingClassLoader != null) { + Thread.currentThread().setContextClassLoader(workingClassLoader); + } + } +} diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ClassLoaderFilter.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ClassLoaderFilter.java index 99bae400fa..c4dd01ccdd 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ClassLoaderFilter.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ClassLoaderFilter.java @@ -18,35 +18,59 @@ package org.apache.dubbo.rpc.filter; import org.apache.dubbo.common.constants.CommonConstants; import org.apache.dubbo.common.extension.Activate; +import org.apache.dubbo.rpc.BaseFilter; import org.apache.dubbo.rpc.Filter; import org.apache.dubbo.rpc.Invocation; import org.apache.dubbo.rpc.Invoker; import org.apache.dubbo.rpc.Result; import org.apache.dubbo.rpc.RpcException; +import static org.apache.dubbo.common.constants.CommonConstants.STAGED_CLASSLOADER_KEY; +import static org.apache.dubbo.common.constants.CommonConstants.WORKING_CLASSLOADER_KEY; + /** * Set the current execution thread class loader to service interface's class loader. */ @Activate(group = CommonConstants.PROVIDER, order = -30000) -public class ClassLoaderFilter implements Filter { +public class ClassLoaderFilter implements Filter, BaseFilter.Listener { @Override public Result invoke(Invoker invoker, Invocation invocation) throws RpcException { - ClassLoader ocl = Thread.currentThread().getContextClassLoader(); - ClassLoader newClassLoader; + ClassLoader stagedClassLoader = Thread.currentThread().getContextClassLoader(); + ClassLoader effectiveClassLoader; if (invocation.getServiceModel() != null) { - newClassLoader = invocation.getServiceModel().getClassLoader(); + effectiveClassLoader = invocation.getServiceModel().getClassLoader(); } else { - newClassLoader = invoker.getClass().getClassLoader(); + effectiveClassLoader = invoker.getClass().getClassLoader(); } - if (newClassLoader != null) { - Thread.currentThread().setContextClassLoader(newClassLoader); + + if (effectiveClassLoader != null) { + invocation.put(STAGED_CLASSLOADER_KEY, stagedClassLoader); + invocation.put(WORKING_CLASSLOADER_KEY, effectiveClassLoader); + + Thread.currentThread().setContextClassLoader(effectiveClassLoader); } try { return invoker.invoke(invocation); } finally { - Thread.currentThread().setContextClassLoader(ocl); + Thread.currentThread().setContextClassLoader(stagedClassLoader); } } + @Override + public void onResponse(Result appResponse, Invoker invoker, Invocation invocation) { + resetClassLoader(invoker, invocation); + } + + @Override + public void onError(Throwable t, Invoker invoker, Invocation invocation) { + resetClassLoader(invoker, invocation); + } + + private void resetClassLoader(Invoker invoker, Invocation invocation) { + ClassLoader stagedClassLoader = (ClassLoader) invocation.get(STAGED_CLASSLOADER_KEY); + if (stagedClassLoader != null) { + Thread.currentThread().setContextClassLoader(stagedClassLoader); + } + } } diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ContextFilter.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ContextFilter.java index 08a1286ead..3d6aabefe9 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ContextFilter.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ContextFilter.java @@ -126,27 +126,29 @@ public class ContextFilter implements Filter, Filter.Listener { ((RpcInvocation) invocation).setInvoker(invoker); } - try { - context.clearAfterEachInvoke(false); - return invoker.invoke(invocation); - } finally { - context.clearAfterEachInvoke(true); - RpcContext.removeServerAttachment(); - RpcContext.removeServiceContext(); - // IMPORTANT! For async scenario, we must remove context from current thread, so we always create a new RpcContext for the next invoke for the same thread. - RpcContext.getClientAttachment().removeAttachment(TIME_COUNTDOWN_KEY); - RpcContext.removeServerContext(); - } + context.clearAfterEachInvoke(false); + + return invoker.invoke(invocation); } @Override public void onResponse(Result appResponse, Invoker invoker, Invocation invocation) { // pass attachments to result appResponse.addObjectAttachments(RpcContext.getServerContext().getObjectAttachments()); + removeContext(); } @Override public void onError(Throwable t, Invoker invoker, Invocation invocation) { + removeContext(); + } + private void removeContext() { + RpcContext.getServerAttachment().clearAfterEachInvoke(true); // TODO, not necessary anymore + + RpcContext.removeServerAttachment(); + RpcContext.removeClientAttachment(); + RpcContext.removeServiceContext(); + RpcContext.removeServerContext(); } } diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/proxy/AbstractProxyInvoker.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/proxy/AbstractProxyInvoker.java index 2230f50796..84dfd338f0 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/proxy/AbstractProxyInvoker.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/proxy/AbstractProxyInvoker.java @@ -32,6 +32,8 @@ import java.lang.reflect.InvocationTargetException; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; +import static org.apache.dubbo.common.constants.CommonConstants.PROVIDER_ASYNC_KEY; + /** * This Invoker works on provider side, delegates RPC to interface implementation. */ @@ -82,7 +84,7 @@ public abstract class AbstractProxyInvoker implements Invoker { public Result invoke(Invocation invocation) throws RpcException { try { Object value = doInvoke(proxy, invocation.getMethodName(), invocation.getParameterTypes(), invocation.getArguments()); - CompletableFuture future = wrapWithFuture(value); + CompletableFuture future = wrapWithFuture(value, invocation); CompletableFuture appResponseFuture = future.handle((obj, t) -> { AppResponse result = new AppResponse(invocation); if (t != null) { @@ -107,11 +109,13 @@ public abstract class AbstractProxyInvoker implements Invoker { } } - private CompletableFuture wrapWithFuture(Object value) { - if (RpcContext.getServiceContext().isAsyncStarted()) { - return ((AsyncContextImpl)(RpcContext.getServiceContext().getAsyncContext())).getInternalFuture(); - } else if (value instanceof CompletableFuture) { + private CompletableFuture wrapWithFuture(Object value, Invocation invocation) { + if (value instanceof CompletableFuture) { + invocation.put(PROVIDER_ASYNC_KEY, Boolean.TRUE); return (CompletableFuture) value; + } else if (RpcContext.getServiceContext().isAsyncStarted()) { + invocation.put(PROVIDER_ASYNC_KEY, Boolean.TRUE); + return ((AsyncContextImpl) (RpcContext.getServiceContext().getAsyncContext())).getInternalFuture(); } return CompletableFuture.completedFuture(value); } diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.Filter b/dubbo-rpc/dubbo-rpc-api/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.Filter index 7c526c25ae..34d56b4358 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.Filter +++ b/dubbo-rpc/dubbo-rpc-api/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.Filter @@ -4,10 +4,11 @@ genericimpl=org.apache.dubbo.rpc.filter.GenericImplFilter token=org.apache.dubbo.rpc.filter.TokenFilter accesslog=org.apache.dubbo.rpc.filter.AccessLogFilter classloader=org.apache.dubbo.rpc.filter.ClassLoaderFilter +classloader-callback=org.apache.dubbo.rpc.filter.ClassLoaderCallbackFilter context=org.apache.dubbo.rpc.filter.ContextFilter exception=org.apache.dubbo.rpc.filter.ExceptionFilter executelimit=org.apache.dubbo.rpc.filter.ExecuteLimitFilter deprecated=org.apache.dubbo.rpc.filter.DeprecatedFilter compatible=org.apache.dubbo.rpc.filter.CompatibleFilter timeout=org.apache.dubbo.rpc.filter.TimeoutFilter -tps=org.apache.dubbo.rpc.filter.TpsLimitFilter \ No newline at end of file +tps=org.apache.dubbo.rpc.filter.TpsLimitFilter diff --git a/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/RpcContextTest.java b/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/RpcContextTest.java index 4f1e6e9448..a7dbd8f293 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/RpcContextTest.java +++ b/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/RpcContextTest.java @@ -202,4 +202,9 @@ public class RpcContextTest { rpcContext.setObjectAttachments(map); Assertions.assertEquals(map, rpcContext.getObjectAttachments()); } + + @Test + public void testRestore() { + + } } diff --git a/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/filter/CompatibleFilterFilterTest.java b/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/filter/CompatibleFilterFilterTest.java index 7750195c29..9a5152bc6a 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/filter/CompatibleFilterFilterTest.java +++ b/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/filter/CompatibleFilterFilterTest.java @@ -22,6 +22,7 @@ import org.apache.dubbo.rpc.AsyncRpcResult; import org.apache.dubbo.rpc.Invocation; import org.apache.dubbo.rpc.Invoker; import org.apache.dubbo.rpc.Result; +import org.apache.dubbo.rpc.RpcInvocation; import org.apache.dubbo.rpc.support.DemoService; import org.apache.dubbo.rpc.support.Type; @@ -49,7 +50,7 @@ public class CompatibleFilterFilterTest { @Test public void testInvokerGeneric() { - invocation = mock(Invocation.class); + invocation = mock(RpcInvocation.class); given(invocation.getMethodName()).willReturn("$enumlength"); given(invocation.getParameterTypes()).willReturn(new Class[]{Enum.class}); given(invocation.getArguments()).willReturn(new Object[]{"hello"}); @@ -69,7 +70,7 @@ public class CompatibleFilterFilterTest { @Test public void testResultHasException() { - invocation = mock(Invocation.class); + invocation = mock(RpcInvocation.class); given(invocation.getMethodName()).willReturn("enumlength"); given(invocation.getParameterTypes()).willReturn(new Class[]{Enum.class}); given(invocation.getArguments()).willReturn(new Object[]{"hello"}); @@ -90,7 +91,7 @@ public class CompatibleFilterFilterTest { @Test public void testInvokerJsonPojoSerialization() throws Exception { - invocation = mock(Invocation.class); + invocation = mock(RpcInvocation.class); given(invocation.getMethodName()).willReturn("enumlength"); given(invocation.getParameterTypes()).willReturn(new Class[]{Type[].class}); given(invocation.getArguments()).willReturn(new Object[]{"hello"}); @@ -100,7 +101,8 @@ public class CompatibleFilterFilterTest { given(invoker.getInterface()).willReturn(DemoService.class); AppResponse result = new AppResponse(); result.setValue("High"); - given(invoker.invoke(invocation)).willReturn(AsyncRpcResult.newDefaultAsyncResult(result, invocation)); + AsyncRpcResult defaultAsyncResult = AsyncRpcResult.newDefaultAsyncResult(result, invocation); + given(invoker.invoke(invocation)).willReturn(defaultAsyncResult); URL url = URL.valueOf("test://test:11/test?group=dubbo&version=1.1&serialization=json"); given(invoker.getUrl()).willReturn(url); @@ -112,7 +114,7 @@ public class CompatibleFilterFilterTest { @Test public void testInvokerNonJsonEnumSerialization() throws Exception { - invocation = mock(Invocation.class); + invocation = mock(RpcInvocation.class); given(invocation.getMethodName()).willReturn("enumlength"); given(invocation.getParameterTypes()).willReturn(new Class[]{Type[].class}); given(invocation.getArguments()).willReturn(new Object[]{"hello"}); @@ -122,7 +124,8 @@ public class CompatibleFilterFilterTest { given(invoker.getInterface()).willReturn(DemoService.class); AppResponse result = new AppResponse(); result.setValue("High"); - given(invoker.invoke(invocation)).willReturn(AsyncRpcResult.newDefaultAsyncResult(result, invocation)); + AsyncRpcResult defaultAsyncResult = AsyncRpcResult.newDefaultAsyncResult(result, invocation); + given(invoker.invoke(invocation)).willReturn(defaultAsyncResult); URL url = URL.valueOf("test://test:11/test?group=dubbo&version=1.1"); given(invoker.getUrl()).willReturn(url); @@ -134,7 +137,7 @@ public class CompatibleFilterFilterTest { @Test public void testInvokerNonJsonNonPojoSerialization() { - invocation = mock(Invocation.class); + invocation = mock(RpcInvocation.class); given(invocation.getMethodName()).willReturn("echo"); given(invocation.getParameterTypes()).willReturn(new Class[]{String.class}); given(invocation.getArguments()).willReturn(new Object[]{"hello"}); @@ -154,7 +157,7 @@ public class CompatibleFilterFilterTest { @Test public void testInvokerNonJsonPojoSerialization() { - invocation = mock(Invocation.class); + invocation = mock(RpcInvocation.class); given(invocation.getMethodName()).willReturn("echo"); given(invocation.getParameterTypes()).willReturn(new Class[]{String.class}); given(invocation.getArguments()).willReturn(new Object[]{"hello"}); diff --git a/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/filter/ContextFilterTest.java b/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/filter/ContextFilterTest.java index 84b6a686bf..a37f87a98a 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/filter/ContextFilterTest.java +++ b/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/filter/ContextFilterTest.java @@ -29,7 +29,7 @@ import org.apache.dubbo.rpc.support.MyInvoker; import org.junit.jupiter.api.Test; -import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.mockito.BDDMockito.given; import static org.mockito.Mockito.mock; @@ -62,7 +62,7 @@ public class ContextFilterTest { given(invoker.getUrl()).willReturn(url); contextFilter.invoke(invoker, invocation); - assertNull(RpcContext.getServiceContext().getInvoker()); + assertNotNull(RpcContext.getServiceContext().getInvoker()); } @Test @@ -71,6 +71,6 @@ public class ContextFilterTest { Invoker invoker = new MyInvoker(url); Invocation invocation = new MockInvocation(); Result result = contextFilter.invoke(invoker, invocation); - assertNull(RpcContext.getServiceContext().getInvoker()); + assertNotNull(RpcContext.getServiceContext().getInvoker()); } } diff --git a/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/filter/EchoFilterTest.java b/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/filter/EchoFilterTest.java index 76ee02480c..b31d22595a 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/filter/EchoFilterTest.java +++ b/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/filter/EchoFilterTest.java @@ -22,6 +22,7 @@ import org.apache.dubbo.rpc.Filter; import org.apache.dubbo.rpc.Invocation; import org.apache.dubbo.rpc.Invoker; import org.apache.dubbo.rpc.Result; +import org.apache.dubbo.rpc.RpcInvocation; import org.apache.dubbo.rpc.support.DemoService; import org.junit.jupiter.api.Test; @@ -37,7 +38,7 @@ public class EchoFilterTest { @SuppressWarnings("unchecked") @Test public void testEcho() { - Invocation invocation = mock(Invocation.class); + Invocation invocation = mock(RpcInvocation.class); given(invocation.getMethodName()).willReturn("$echo"); given(invocation.getParameterTypes()).willReturn(new Class[]{Enum.class}); given(invocation.getArguments()).willReturn(new Object[]{"hello"}); diff --git a/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/filter/TimeoutFilterTest.java b/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/filter/TimeoutFilterTest.java index be85365333..256427377c 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/filter/TimeoutFilterTest.java +++ b/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/filter/TimeoutFilterTest.java @@ -21,6 +21,7 @@ import org.apache.dubbo.rpc.AppResponse; import org.apache.dubbo.rpc.Invocation; import org.apache.dubbo.rpc.Invoker; import org.apache.dubbo.rpc.Result; +import org.apache.dubbo.rpc.RpcInvocation; import org.apache.dubbo.rpc.support.BlockMyInvoker; import org.junit.jupiter.api.Assertions; @@ -56,7 +57,7 @@ public class TimeoutFilterTest { URL url = URL.valueOf("test://test:11/test?accesslog=true&group=dubbo&version=1.1&timeout=" + timeout); Invoker invoker = new BlockMyInvoker(url, (timeout + 100)); - Invocation invocation = Mockito.mock(Invocation.class); + Invocation invocation = Mockito.mock(RpcInvocation.class); when(invocation.getMethodName()).thenReturn("testInvokeWithTimeout"); Result result = timeoutFilter.invoke(invoker, invocation); diff --git a/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/support/MockInvocation.java b/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/support/MockInvocation.java index 2c8a618429..ae8a69cc5d 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/support/MockInvocation.java +++ b/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/support/MockInvocation.java @@ -17,8 +17,8 @@ package org.apache.dubbo.rpc.support; import org.apache.dubbo.rpc.AttachmentsAdapter; -import org.apache.dubbo.rpc.Invocation; import org.apache.dubbo.rpc.Invoker; +import org.apache.dubbo.rpc.RpcInvocation; import org.apache.dubbo.rpc.model.ServiceModel; import java.util.HashMap; @@ -34,7 +34,7 @@ import static org.apache.dubbo.rpc.Constants.TOKEN_KEY; /** * MockInvocation.java */ -public class MockInvocation implements Invocation { +public class MockInvocation extends RpcInvocation { private Map attachments; diff --git a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboProtocol.java b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboProtocol.java index 50110d723a..eed8192fff 100644 --- a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboProtocol.java +++ b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboProtocol.java @@ -607,7 +607,7 @@ public class DubboProtocol extends AbstractProtocol { */ private ReferenceCountExchangeClient buildReferenceCountExchangeClient(URL url) { ExchangeClient exchangeClient = initClient(url); - ReferenceCountExchangeClient client = new ReferenceCountExchangeClient(exchangeClient); + ReferenceCountExchangeClient client = new ReferenceCountExchangeClient(exchangeClient, DubboCodec.NAME); // read configs int shutdownTimeout = ConfigurationUtils.getServerShutdownTimeout(url.getScopeModel()); client.setShutdownWaitTime(shutdownTimeout); @@ -640,7 +640,7 @@ public class DubboProtocol extends AbstractProtocol { try { // connection should be lazy if (url.getParameter(LAZY_CONNECT_KEY, false)) { - client = new LazyConnectExchangeClient(url, requestHandler); + client = new LazyConnectExchangeClient(url, requestHandler, DubboCodec.NAME, url.getParameters()); } else { client = Exchangers.connect(url, requestHandler); diff --git a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/LazyConnectExchangeClient.java b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/LazyConnectExchangeClient.java index f31144a726..a5bb837461 100644 --- a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/LazyConnectExchangeClient.java +++ b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/LazyConnectExchangeClient.java @@ -29,6 +29,7 @@ import org.apache.dubbo.remoting.exchange.ExchangeHandler; import org.apache.dubbo.remoting.exchange.Exchangers; import java.net.InetSocketAddress; +import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutorService; import java.util.concurrent.atomic.AtomicLong; @@ -60,9 +61,11 @@ final class LazyConnectExchangeClient implements ExchangeClient { private volatile ExchangeClient client; private final AtomicLong warningCount = new AtomicLong(0); - public LazyConnectExchangeClient(URL url, ExchangeHandler requestHandler) { + public LazyConnectExchangeClient(URL url, ExchangeHandler requestHandler, String codec, Map parameters) { // lazy connect, need set send.reconnect = true, to avoid channel bad status. - this.url = new ServiceConfigURL(url.getProtocol(), url.getUsername(), url.getPassword(), url.getHost(), url.getPort(), url.getPath(), url.getParameters()) + // Parameters like 'username', 'password' and 'path' are set but will not be used in following processes. + // The most important parameters here are 'host', port', 'parameters' and 'codec', 'codec' can also be extracted from 'parameters' + this.url = new ServiceConfigURL(codec, url.getUsername(), url.getPassword(), url.getHost(), url.getPort(), url.getPath(), parameters) .addParameter(SEND_RECONNECT_KEY, Boolean.TRUE.toString()); this.requestHandler = requestHandler; this.initialState = url.getParameter(LAZY_CONNECT_INITIAL_STATE_KEY, DEFAULT_LAZY_CONNECT_INITIAL_STATE); diff --git a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/ReferenceCountExchangeClient.java b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/ReferenceCountExchangeClient.java index a2ff32b3fb..9fc09f9788 100644 --- a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/ReferenceCountExchangeClient.java +++ b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/ReferenceCountExchangeClient.java @@ -27,6 +27,7 @@ import org.apache.dubbo.remoting.exchange.ExchangeClient; import org.apache.dubbo.remoting.exchange.ExchangeHandler; import java.net.InetSocketAddress; +import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutorService; import java.util.concurrent.atomic.AtomicInteger; @@ -43,16 +44,20 @@ final class ReferenceCountExchangeClient implements ExchangeClient { private final static Logger logger = LoggerFactory.getLogger(ReferenceCountExchangeClient.class); private final URL url; + private final String codec; + private final Map attributes; private final AtomicInteger referenceCount = new AtomicInteger(0); private final AtomicInteger disconnectCount = new AtomicInteger(0); private final Integer warningPeriod = 50; private ExchangeClient client; private int shutdownWaitTime = DEFAULT_SERVER_SHUTDOWN_TIMEOUT; - public ReferenceCountExchangeClient(ExchangeClient client) { + public ReferenceCountExchangeClient(ExchangeClient client, String codec) { this.client = client; - referenceCount.incrementAndGet(); + this.referenceCount.incrementAndGet(); this.url = client.getUrl(); + this.codec = codec; + this.attributes = url.getParameters(); } @Override @@ -210,10 +215,16 @@ final class ReferenceCountExchangeClient implements ExchangeClient { if (!(client instanceof LazyConnectExchangeClient)) { // this is a defensive operation to avoid client is closed by accident, the initial state of the client is false URL lazyUrl = url.addParameter(LAZY_CONNECT_INITIAL_STATE_KEY, Boolean.TRUE) - //.addParameter(RECONNECT_KEY, Boolean.FALSE) - .addParameter(SEND_RECONNECT_KEY, Boolean.TRUE.toString()); + //.addParameter(RECONNECT_KEY, Boolean.FALSE) + .addParameter(SEND_RECONNECT_KEY, Boolean.TRUE.toString()); //.addParameter(LazyConnectExchangeClient.REQUEST_WITH_WARNING_KEY, true); - client = new LazyConnectExchangeClient(lazyUrl, client.getExchangeHandler()); + + // uncomment this snippet when replacing lazyUrl in the futrue +// Map lazyAttributes = new HashMap<>(attributes); +// lazyAttributes.put(LAZY_CONNECT_INITIAL_STATE_KEY, Boolean.TRUE.toString()); +// lazyAttributes.put(SEND_RECONNECT_KEY, Boolean.TRUE.toString()); + + client = new LazyConnectExchangeClient(lazyUrl, client.getExchangeHandler(), codec, attributes); } } diff --git a/dubbo-rpc/dubbo-rpc-dubbo/src/test/java/org/apache/dubbo/rpc/protocol/dubbo/DubboProtocolTest.java b/dubbo-rpc/dubbo-rpc-dubbo/src/test/java/org/apache/dubbo/rpc/protocol/dubbo/DubboProtocolTest.java index 14ac6de942..d9760e4bac 100644 --- a/dubbo-rpc/dubbo-rpc-dubbo/src/test/java/org/apache/dubbo/rpc/protocol/dubbo/DubboProtocolTest.java +++ b/dubbo-rpc/dubbo-rpc-dubbo/src/test/java/org/apache/dubbo/rpc/protocol/dubbo/DubboProtocolTest.java @@ -78,7 +78,7 @@ public class DubboProtocolTest { int port = NetUtils.getAvailablePort(); protocol.export(proxy.getInvoker(service, DemoService.class, URL.valueOf("dubbo://127.0.0.1:" + port + "/" + DemoService.class.getName() + "?codec=exchange"))); service = proxy.getProxy(protocol.refer(DemoService.class, URL.valueOf("dubbo://127.0.0.1:" + port + "/" + DemoService.class.getName() + "?codec=exchange").addParameter("timeout", - 3000L))); + 3000L))); assertEquals(service.getSize(new String[]{"", "", ""}), 3); } @@ -88,7 +88,7 @@ public class DubboProtocolTest { int port = NetUtils.getAvailablePort(); protocol.export(proxy.getInvoker(service, DemoService.class, URL.valueOf("dubbo://127.0.0.1:" + port + "/" + DemoService.class.getName()))); service = proxy.getProxy(protocol.refer(DemoService.class, URL.valueOf("dubbo://127.0.0.1:" + port + "/" + DemoService.class.getName()).addParameter("timeout", - 3000L))); + 3000L))); assertEquals(service.enumlength(new Type[]{}), Type.Lower); assertEquals(service.getSize(null), -1); assertEquals(service.getSize(new String[]{"", "", ""}), 3); @@ -100,7 +100,7 @@ public class DubboProtocolTest { service.invoke("dubbo://127.0.0.1:" + port + "/" + DemoService.class.getName() + "", "invoke"); service = proxy.getProxy(protocol.refer(DemoService.class, URL.valueOf("dubbo://127.0.0.1:" + port + "/" + DemoService.class.getName() + "?client=netty").addParameter("timeout", - 3000L))); + 3000L))); // test netty client StringBuffer buf = new StringBuffer(); for (int i = 0; i < 1024 * 32 + 32; i++) @@ -109,7 +109,7 @@ public class DubboProtocolTest { // cast to EchoService EchoService echo = proxy.getProxy(protocol.refer(EchoService.class, URL.valueOf("dubbo://127.0.0.1:" + port + "/" + DemoService.class.getName() + "?client=netty").addParameter("timeout", - 3000L))); + 3000L))); assertEquals(echo.$echo(buf.toString()), buf.toString()); assertEquals(echo.$echo("test"), "test"); assertEquals(echo.$echo("abcdefg"), "abcdefg"); @@ -168,9 +168,11 @@ public class DubboProtocolTest { ApplicationModel.defaultModel().getDefaultModule().getServiceRepository().registerService(RemoteService.class); int port = NetUtils.getAvailablePort(); - protocol.export(proxy.getInvoker(remote, RemoteService.class, URL.valueOf("dubbo://127.0.0.1:" + port + "/" + RemoteService.class.getName()))); - remote = proxy.getProxy(protocol.refer(RemoteService.class, URL.valueOf("dubbo://127.0.0.1:" + port + "/" + RemoteService.class.getName()).addParameter("timeout", - 3000L))); + URL url = URL.valueOf("dubbo://127.0.0.1:" + port + "/" + RemoteService.class.getName()).addParameter("timeout", + 3000L); + url = url.setScopeModel(ApplicationModel.defaultModel().getDefaultModule()); + protocol.export(proxy.getInvoker(remote, RemoteService.class, url)); + remote = proxy.getProxy(protocol.refer(RemoteService.class, url)); // service.sayHello("world"); @@ -189,7 +191,7 @@ public class DubboProtocolTest { int port = NetUtils.getAvailablePort(); protocol.export(proxy.getInvoker(service, DemoService.class, URL.valueOf("dubbo://127.0.0.1:" + port + "/" + DemoService.class.getName() + "?codec=exchange"))); service = proxy.getProxy(protocol.refer(DemoService.class, URL.valueOf("dubbo://127.0.0.1:" + port + "/" + DemoService.class.getName() + "?codec=exchange").addParameter("timeout", - 3000L))); + 3000L))); long start = System.currentTimeMillis(); for (int i = 0; i < 100; i++) service.getSize(new String[]{"", "", ""}); @@ -202,7 +204,7 @@ public class DubboProtocolTest { int port = NetUtils.getAvailablePort(); protocol.export(proxy.getInvoker(service, DemoService.class, URL.valueOf("dubbo://127.0.0.1:" + port + "/" + DemoService.class.getName() + "?codec=exchange"))); service = proxy.getProxy(protocol.refer(DemoService.class, URL.valueOf("dubbo://127.0.0.1:" + port + "/" + DemoService.class.getName() + "?codec=exchange").addParameter("timeout", - 3000L))); + 3000L))); try { service.nonSerializedParameter(new NonSerialized()); Assertions.fail(); @@ -217,7 +219,7 @@ public class DubboProtocolTest { int port = NetUtils.getAvailablePort(); protocol.export(proxy.getInvoker(service, DemoService.class, URL.valueOf("dubbo://127.0.0.1:" + port + "/" + DemoService.class.getName() + "?codec=exchange"))); service = proxy.getProxy(protocol.refer(DemoService.class, URL.valueOf("dubbo://127.0.0.1:" + port + "/" + DemoService.class.getName() + "?codec=exchange").addParameter("timeout", - 3000L))); + 3000L))); try { service.returnNonSerialized(); Assertions.fail(); @@ -231,7 +233,8 @@ public class DubboProtocolTest { DemoService service = new DemoServiceImpl(); int port = NetUtils.getAvailablePort(); URL url = URL.valueOf("dubbo://127.0.0.1:" + port + "/" + DemoService.class.getName() + "?codec=exchange").addParameter("timeout", - 3000L).addParameter("application", "consumer"); + 3000L).addParameter("application", "consumer"); + url = url.setScopeModel(ApplicationModel.defaultModel().getDefaultModule()); protocol.export(proxy.getInvoker(service, DemoService.class, url)); Invoker invoker = protocol.refer(DemoService.class, url); @@ -254,10 +257,10 @@ public class DubboProtocolTest { DemoService service = new DemoServiceImpl(); int port = NetUtils.getAvailablePort(); protocol.export(proxy.getInvoker(service, DemoService.class, - URL.valueOf("dubbo://127.0.0.1:" + port + "/" + DemoService.class.getName()).addParameter("payload", 10 * 1024))); + URL.valueOf("dubbo://127.0.0.1:" + port + "/" + DemoService.class.getName()).addParameter("payload", 10 * 1024))); service = proxy.getProxy(protocol.refer(DemoService.class, - URL.valueOf("dubbo://127.0.0.1:" + port + "/" + DemoService.class.getName()).addParameter("timeout", - 6000L).addParameter("payload", 160))); + URL.valueOf("dubbo://127.0.0.1:" + port + "/" + DemoService.class.getName()).addParameter("timeout", + 6000L).addParameter("payload", 160))); try { service.download(300); Assertions.fail(); diff --git a/dubbo-rpc/dubbo-rpc-dubbo/src/test/java/org/apache/dubbo/rpc/protocol/dubbo/RpcFilterTest.java b/dubbo-rpc/dubbo-rpc-dubbo/src/test/java/org/apache/dubbo/rpc/protocol/dubbo/RpcFilterTest.java index d34d21f375..fc891083ce 100644 --- a/dubbo-rpc/dubbo-rpc-dubbo/src/test/java/org/apache/dubbo/rpc/protocol/dubbo/RpcFilterTest.java +++ b/dubbo-rpc/dubbo-rpc-dubbo/src/test/java/org/apache/dubbo/rpc/protocol/dubbo/RpcFilterTest.java @@ -45,6 +45,7 @@ public class RpcFilterTest { int port = NetUtils.getAvailablePort(); URL url = URL.valueOf("dubbo://127.0.0.1:" + port + "/org.apache.dubbo.rpc.protocol.dubbo.support.DemoService?service.filter=echo"); ApplicationModel.defaultModel().getDefaultModule().getServiceRepository().registerService(DemoService.class); + url = url.setScopeModel(ApplicationModel.defaultModel().getDefaultModule()); protocol.export(proxy.getInvoker(service, DemoService.class, url)); service = proxy.getProxy(protocol.refer(DemoService.class, url)); Assertions.assertEquals("123", service.echo("123")); diff --git a/dubbo-spring-boot/dubbo-spring-boot-actuator/pom.xml b/dubbo-spring-boot/dubbo-spring-boot-actuator/pom.xml index c0ec86995a..b39fee024e 100644 --- a/dubbo-spring-boot/dubbo-spring-boot-actuator/pom.xml +++ b/dubbo-spring-boot/dubbo-spring-boot-actuator/pom.xml @@ -43,6 +43,19 @@ org.springframework.boot spring-boot-starter-web true + + + + org.apache.logging.log4j + log4j-api + + + + + org.apache.logging.log4j + log4j-api + ${log4j2_version} + true @@ -111,6 +124,13 @@ org.springframework.boot spring-boot-starter-test test + + + + org.apache.logging.log4j + log4j-api + + diff --git a/dubbo-spring-boot/dubbo-spring-boot-autoconfigure/pom.xml b/dubbo-spring-boot/dubbo-spring-boot-autoconfigure/pom.xml index 80a063bfd9..30ddd03b35 100644 --- a/dubbo-spring-boot/dubbo-spring-boot-autoconfigure/pom.xml +++ b/dubbo-spring-boot/dubbo-spring-boot-autoconfigure/pom.xml @@ -51,6 +51,13 @@ org.springframework.boot spring-boot-starter-logging true + + + + org.apache.logging.log4j + log4j-api + + @@ -79,6 +86,13 @@ org.springframework.boot spring-boot-starter-test test + + + + org.apache.logging.log4j + log4j-api + + diff --git a/dubbo-spring-boot/dubbo-spring-boot-compatible/actuator/pom.xml b/dubbo-spring-boot/dubbo-spring-boot-compatible/actuator/pom.xml index 23af65866a..a1a84ec476 100644 --- a/dubbo-spring-boot/dubbo-spring-boot-compatible/actuator/pom.xml +++ b/dubbo-spring-boot/dubbo-spring-boot-compatible/actuator/pom.xml @@ -101,6 +101,13 @@ org.springframework.boot spring-boot-starter-test test + + + + org.apache.logging.log4j + log4j-api + + diff --git a/dubbo-spring-boot/dubbo-spring-boot-compatible/autoconfigure/pom.xml b/dubbo-spring-boot/dubbo-spring-boot-compatible/autoconfigure/pom.xml index e99f6b7f65..dc940e5c44 100644 --- a/dubbo-spring-boot/dubbo-spring-boot-compatible/autoconfigure/pom.xml +++ b/dubbo-spring-boot/dubbo-spring-boot-compatible/autoconfigure/pom.xml @@ -48,6 +48,13 @@ org.springframework.boot spring-boot-starter-logging true + + + + org.apache.logging.log4j + log4j-api + + @@ -82,6 +89,13 @@ org.springframework.boot spring-boot-starter-test test + + + + org.apache.logging.log4j + log4j-api + + diff --git a/dubbo-spring-boot/dubbo-spring-boot-starter/pom.xml b/dubbo-spring-boot/dubbo-spring-boot-starter/pom.xml index 763795bfb4..cb1ce7ac7f 100644 --- a/dubbo-spring-boot/dubbo-spring-boot-starter/pom.xml +++ b/dubbo-spring-boot/dubbo-spring-boot-starter/pom.xml @@ -37,6 +37,19 @@ org.springframework.boot spring-boot-starter true + + + + org.apache.logging.log4j + log4j-api + + + + + org.apache.logging.log4j + log4j-api + ${log4j2_version} + true diff --git a/dubbo-spring-boot/pom.xml b/dubbo-spring-boot/pom.xml index 2f52e8a22f..9f1de103ed 100644 --- a/dubbo-spring-boot/pom.xml +++ b/dubbo-spring-boot/pom.xml @@ -41,6 +41,8 @@ 2.3.1.RELEASE ${revision} + + 2.17.0 diff --git a/pom.xml b/pom.xml index e202b5b738..e16b6037ea 100644 --- a/pom.xml +++ b/pom.xml @@ -129,7 +129,7 @@ true true true - 3.0.5-SNAPSHOT + 3.0.5 @@ -379,7 +379,7 @@ release - 2.11.1 + 2.17.0 false