diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/AbstractClusterInvoker.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/AbstractClusterInvoker.java index 861f5ab789..745e602f1f 100644 --- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/AbstractClusterInvoker.java +++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/AbstractClusterInvoker.java @@ -379,7 +379,7 @@ public abstract class AbstractClusterInvoker implements ClusterInvoker { if (ProfilerSwitch.isEnableSimpleProfiler()) { InvocationProfilerUtils.enterProfiler(invocation, "Invoker invoke. Target Address: " + invoker.getUrl().getAddress()); } - invocation.addInvokedInvoker(invoker); + setRemote(invoker, invocation); result = invoker.invoke(invocation); } finally { clearContext(originInvoker); @@ -388,6 +388,17 @@ public abstract class AbstractClusterInvoker implements ClusterInvoker { return result; } + /** + * Set the remoteAddress and remoteApplicationName so that filter can get them. + * + */ + private void setRemote(Invoker invoker, Invocation invocation) { + invocation.addInvokedInvoker(invoker); + RpcServiceContext serviceContext = RpcContext.getServiceContext(); + serviceContext.setRemoteAddress(invoker.getUrl().toInetSocketAddress()); + serviceContext.setRemoteApplicationName(invoker.getUrl().getRemoteApplication()); + } + /** * When using a thread pool to fork a child thread, ThreadLocal cannot be passed. * In this scenario, please use the invokeWithContextAsync method. 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 dae8a0ca03..043cdd9e78 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 @@ -60,6 +60,8 @@ public final class ConfigurationUtils { private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(ConfigurationUtils.class); private static final List securityKey; + private static volatile long expectedShutdownTime = Long.MAX_VALUE; + static { List keys = new LinkedList<>(); keys.add("accesslog"); @@ -113,6 +115,9 @@ public final class ConfigurationUtils { */ @SuppressWarnings("deprecation") public static int getServerShutdownTimeout(ScopeModel scopeModel) { + if (expectedShutdownTime < System.currentTimeMillis()) { + return 1; + } int timeout = DEFAULT_SERVER_SHUTDOWN_TIMEOUT; Configuration configuration = getGlobalConfiguration(scopeModel); String value = StringUtils.trim(configuration.getString(SHUTDOWN_WAIT_KEY)); @@ -133,9 +138,33 @@ public final class ConfigurationUtils { } } } + + if (expectedShutdownTime - System.currentTimeMillis() < timeout) { + return (int) Math.max(1, expectedShutdownTime - System.currentTimeMillis()); + } + return timeout; } + public static int reCalShutdownTime(int expected) { + // already timeout + if (expectedShutdownTime < System.currentTimeMillis()) { + return 1; + } + + if (expectedShutdownTime - System.currentTimeMillis() < expected) { + // the shutdown time rest is less than expected + return (int) Math.max(1, expectedShutdownTime - System.currentTimeMillis()); + } + + // return the expected + return expected; + } + + public static void setExpectedShutdownTime(long expectedShutdownTime) { + ConfigurationUtils.expectedShutdownTime = expectedShutdownTime; + } + public static String getCachedDynamicProperty(ScopeModel realScopeModel, String key, String defaultValue) { ScopeModel scopeModel = getScopeModelOrDefaultApplicationModel(realScopeModel); ConfigurationCache configurationCache = scopeModel.getBeanFactory().getBean(ConfigurationCache.class); diff --git a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/DubboShutdownHook.java b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/DubboShutdownHook.java index def7a7ac74..a8dc78d606 100644 --- a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/DubboShutdownHook.java +++ b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/DubboShutdownHook.java @@ -82,6 +82,9 @@ public class DubboShutdownHook extends Thread { } private void doDestroy() { + int timeout = ConfigurationUtils.getServerShutdownTimeout(applicationModel); + ConfigurationUtils.setExpectedShutdownTime(System.currentTimeMillis() + timeout); + // send readonly for shutdown hook List gracefulShutdowns = GracefulShutdown.getGracefulShutdowns(applicationModel.getFrameworkModel()); for (GracefulShutdown gracefulShutdown : gracefulShutdowns) { @@ -97,7 +100,6 @@ public class DubboShutdownHook extends Thread { } } if (hasModuleBindSpring) { - int timeout = ConfigurationUtils.getServerShutdownTimeout(applicationModel); if (timeout > 0) { long start = System.currentTimeMillis(); /* diff --git a/dubbo-plugin/dubbo-plugin-context/src/main/java/org/apache/dubbo/rpc/cluster/filter/support/ConsumerContextFilter.java b/dubbo-plugin/dubbo-plugin-context/src/main/java/org/apache/dubbo/rpc/cluster/filter/support/ConsumerContextFilter.java index 582578b5e9..a5a6dcf534 100644 --- a/dubbo-plugin/dubbo-plugin-context/src/main/java/org/apache/dubbo/rpc/cluster/filter/support/ConsumerContextFilter.java +++ b/dubbo-plugin/dubbo-plugin-context/src/main/java/org/apache/dubbo/rpc/cluster/filter/support/ConsumerContextFilter.java @@ -62,63 +62,58 @@ public class ConsumerContextFilter implements ClusterFilter, ClusterFilter.Liste @Override public Result invoke(Invoker invoker, Invocation invocation) throws RpcException { - RpcContext.RestoreServiceContext originServiceContext = RpcContext.storeServiceContext(); - try { - RpcContext.getServiceContext() - .setInvoker(invoker) - .setInvocation(invocation) - .setLocalAddress(NetUtils.getLocalHost(), 0); + RpcContext.getServiceContext() + .setInvoker(invoker) + .setInvocation(invocation) + .setLocalAddress(NetUtils.getLocalHost(), 0); - RpcContext context = RpcContext.getClientAttachment(); - context.setAttachment(REMOTE_APPLICATION_KEY, invoker.getUrl().getApplication()); - if (invocation instanceof RpcInvocation) { - ((RpcInvocation) invocation).setInvoker(invoker); - } - - if (CollectionUtils.isNotEmpty(supportedSelectors)) { - for (PenetrateAttachmentSelector supportedSelector : supportedSelectors) { - Map selected = supportedSelector.select(invocation, RpcContext.getClientAttachment(), RpcContext.getServerAttachment()); - if (CollectionUtils.isNotEmptyMap(selected)) { - ((RpcInvocation) invocation).addObjectAttachments(selected); - } - } - } else { - ((RpcInvocation) invocation).addObjectAttachments(RpcContext.getServerAttachment().getObjectAttachments()); - } - Map contextAttachments = RpcContext.getClientAttachment().getObjectAttachments(); - if (CollectionUtils.isNotEmptyMap(contextAttachments)) { - /** - * invocation.addAttachmentsIfAbsent(context){@link RpcInvocation#addAttachmentsIfAbsent(Map)}should not be used here, - * because the {@link RpcContext#setAttachment(String, String)} is passed in the Filter when the call is triggered - * by the built-in retry mechanism of the Dubbo. The attachment to update RpcContext will no longer work, which is - * a mistake in most cases (for example, through Filter to RpcContext output traceId and spanId and other information). - */ - ((RpcInvocation) invocation).addObjectAttachments(contextAttachments); - } - - // pass default timeout set by end user (ReferenceConfig) - Object countDown = RpcContext.getServerAttachment().getObjectAttachment(TIME_COUNTDOWN_KEY); - if (countDown != null) { - String methodName = RpcUtils.getMethodName(invocation); - // When the client has enabled the timeout-countdown function, - // the subsequent calls launched by the Server side will be enabled by default, - // and support to turn off the function on a node to get rid of the timeout control. - if (invoker.getUrl().getMethodParameter(methodName, ENABLE_TIMEOUT_COUNTDOWN_KEY, true)) { - context.setObjectAttachment(TIME_COUNTDOWN_KEY, countDown); - - 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() + "." - + RpcUtils.getMethodName(invocation) + ", terminate directly."), invocation); - } - } - } - RpcContext.removeClientResponseContext(); - return invoker.invoke(invocation); - } finally { - RpcContext.restoreServiceContext(originServiceContext); + RpcContext context = RpcContext.getClientAttachment(); + context.setAttachment(REMOTE_APPLICATION_KEY, invoker.getUrl().getApplication()); + if (invocation instanceof RpcInvocation) { + ((RpcInvocation) invocation).setInvoker(invoker); } + + if (CollectionUtils.isNotEmpty(supportedSelectors)) { + for (PenetrateAttachmentSelector supportedSelector : supportedSelectors) { + Map selected = supportedSelector.select(invocation, RpcContext.getClientAttachment(), RpcContext.getServerAttachment()); + if (CollectionUtils.isNotEmptyMap(selected)) { + ((RpcInvocation) invocation).addObjectAttachments(selected); + } + } + } else { + ((RpcInvocation) invocation).addObjectAttachments(RpcContext.getServerAttachment().getObjectAttachments()); + } + Map contextAttachments = RpcContext.getClientAttachment().getObjectAttachments(); + if (CollectionUtils.isNotEmptyMap(contextAttachments)) { + /** + * invocation.addAttachmentsIfAbsent(context){@link RpcInvocation#addAttachmentsIfAbsent(Map)}should not be used here, + * because the {@link RpcContext#setAttachment(String, String)} is passed in the Filter when the call is triggered + * by the built-in retry mechanism of the Dubbo. The attachment to update RpcContext will no longer work, which is + * a mistake in most cases (for example, through Filter to RpcContext output traceId and spanId and other information). + */ + ((RpcInvocation) invocation).addObjectAttachments(contextAttachments); + } + + // pass default timeout set by end user (ReferenceConfig) + Object countDown = RpcContext.getServerAttachment().getObjectAttachment(TIME_COUNTDOWN_KEY); + if (countDown != null) { + String methodName = RpcUtils.getMethodName(invocation); + // When the client has enabled the timeout-countdown function, + // the subsequent calls launched by the Server side will be enabled by default, + // and support to turn off the function on a node to get rid of the timeout control. + if (invoker.getUrl().getMethodParameter(methodName, ENABLE_TIMEOUT_COUNTDOWN_KEY, true)) { + context.setObjectAttachment(TIME_COUNTDOWN_KEY, countDown); + + 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() + "." + + RpcUtils.getMethodName(invocation) + ", terminate directly."), invocation); + } + } + } + RpcContext.removeClientResponseContext(); + return invoker.invoke(invocation); } @Override diff --git a/dubbo-plugin/dubbo-plugin-loadbalance-adaptive/src/main/java/org/apache/dubbo/rpc/cluster/loadbalance/AdaptiveLoadBalance.java b/dubbo-plugin/dubbo-plugin-loadbalance-adaptive/src/main/java/org/apache/dubbo/rpc/cluster/loadbalance/AdaptiveLoadBalance.java index a406987c97..7f99e371af 100644 --- a/dubbo-plugin/dubbo-plugin-loadbalance-adaptive/src/main/java/org/apache/dubbo/rpc/cluster/loadbalance/AdaptiveLoadBalance.java +++ b/dubbo-plugin/dubbo-plugin-loadbalance-adaptive/src/main/java/org/apache/dubbo/rpc/cluster/loadbalance/AdaptiveLoadBalance.java @@ -110,7 +110,7 @@ public class AdaptiveLoadBalance extends AbstractLoadBalance { private Invoker chooseLowLoadInvoker(Invoker invoker1,Invoker invoker2,Invocation invocation){ int weight1 = getWeight(invoker1, invocation); int weight2 = getWeight(invoker2, invocation); - int timeout1 = getTimeout(invoker2, invocation); + int timeout1 = getTimeout(invoker1, invocation); int timeout2 = getTimeout(invoker2, invocation); long load1 = Double.doubleToLongBits(adaptiveMetrics.getLoad(getServiceKey(invoker1,invocation),weight1,timeout1 )); long load2 = Double.doubleToLongBits(adaptiveMetrics.getLoad(getServiceKey(invoker2,invocation),weight2,timeout2 )); diff --git a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/Constants.java b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/Constants.java index 6a4c67da84..1d00903d0e 100644 --- a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/Constants.java +++ b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/Constants.java @@ -66,6 +66,13 @@ public interface Constants { */ long LEAST_HEARTBEAT_DURATION = 1000; + /** + * the least reconnect during is 60000 ms. + */ + long LEAST_RECONNECT_DURATION = 60000; + + String LEAST_RECONNECT_DURATION_KEY = "dubbo.application.least-reconnect-duration"; + /** * ticks per wheel. */ diff --git a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/HeaderExchangeChannel.java b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/HeaderExchangeChannel.java index 5bb08fa8e2..b4c7bb65a3 100644 --- a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/HeaderExchangeChannel.java +++ b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/HeaderExchangeChannel.java @@ -50,7 +50,7 @@ final class HeaderExchangeChannel implements ExchangeChannel { private final Channel channel; - private final long shutdownTimeout; + private final int shutdownTimeout; private volatile boolean closed = false; @@ -167,7 +167,7 @@ final class HeaderExchangeChannel implements ExchangeChannel { closed = true; try { // graceful close - DefaultFuture.closeChannel(channel, shutdownTimeout); + DefaultFuture.closeChannel(channel, ConfigurationUtils.reCalShutdownTime(shutdownTimeout)); } catch (Exception e) { logger.warn(TRANSPORT_FAILED_CLOSE, "", "", e.getMessage(), e); } diff --git a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/HeaderExchangeClient.java b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/HeaderExchangeClient.java index f776e9e5d7..0adf656604 100644 --- a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/HeaderExchangeClient.java +++ b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/HeaderExchangeClient.java @@ -37,6 +37,8 @@ import java.util.concurrent.TimeUnit; import static org.apache.dubbo.remoting.Constants.HEARTBEAT_CHECK_TICK; import static org.apache.dubbo.remoting.Constants.LEAST_HEARTBEAT_DURATION; +import static org.apache.dubbo.remoting.Constants.LEAST_RECONNECT_DURATION; +import static org.apache.dubbo.remoting.Constants.LEAST_RECONNECT_DURATION_KEY; import static org.apache.dubbo.remoting.Constants.TICKS_PER_WHEEL; import static org.apache.dubbo.remoting.utils.UrlUtils.getHeartbeat; import static org.apache.dubbo.remoting.utils.UrlUtils.getIdleTimeout; @@ -214,7 +216,8 @@ public class HeaderExchangeClient implements ExchangeClient { private void startReconnectTask(URL url) { if (shouldReconnect(url)) { long heartbeatTimeoutTick = calculateLeastDuration(idleTimeout); - reconnectTimerTask = new ReconnectTimerTask(() -> Collections.singleton(this), IDLE_CHECK_TIMER.get(), heartbeatTimeoutTick, idleTimeout); + reconnectTimerTask = new ReconnectTimerTask(() -> Collections.singleton(this), IDLE_CHECK_TIMER.get(), + calculateReconnectDuration(url, heartbeatTimeoutTick), idleTimeout); } } @@ -240,6 +243,11 @@ public class HeaderExchangeClient implements ExchangeClient { } } + private long calculateReconnectDuration(URL url, long tick) { + long leastReconnectDuration = url.getParameter(LEAST_RECONNECT_DURATION_KEY, LEAST_RECONNECT_DURATION); + return Math.max(leastReconnectDuration, tick); + } + private boolean shouldReconnect(URL url) { return url.getParameter(Constants.RECONNECT_KEY, true); } diff --git a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/HeaderExchangeHandler.java b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/HeaderExchangeHandler.java index 2fe6253140..d1fc4cdd1a 100644 --- a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/HeaderExchangeHandler.java +++ b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/HeaderExchangeHandler.java @@ -148,7 +148,7 @@ public class HeaderExchangeHandler implements ChannelHandlerDelegate { if (timeoutObj instanceof Integer) { shutdownTimeout = (Integer) timeoutObj; } - DefaultFuture.closeChannel(channel, shutdownTimeout); + DefaultFuture.closeChannel(channel, ConfigurationUtils.reCalShutdownTime(shutdownTimeout)); HeaderExchangeChannel.removeChannel(channel); } } 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 f4229aa56d..37183b94cb 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 @@ -29,11 +29,13 @@ import org.apache.dubbo.remoting.exchange.Exchangers; import org.apache.dubbo.remoting.exchange.support.ExchangeHandlerAdapter; import org.apache.dubbo.rpc.model.ApplicationModel; import org.apache.dubbo.rpc.model.FrameworkModel; + import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import static org.apache.dubbo.common.constants.CommonConstants.EXECUTOR_MANAGEMENT_MODE_DEFAULT; +import static org.apache.dubbo.remoting.Constants.LEAST_RECONNECT_DURATION_KEY; /** * Client reconnect test @@ -76,7 +78,7 @@ class ClientReconnectTest { public Client startClient(int port, int heartbeat) throws RemotingException { URL url = URL.valueOf("exchange://127.0.0.1:" + port + "/client.reconnect.test?check=false&codec=exchange&client=netty3&" + - Constants.HEARTBEAT_KEY + "=" + heartbeat); + Constants.HEARTBEAT_KEY + "=" + heartbeat + "&" + LEAST_RECONNECT_DURATION_KEY + "=0"); ApplicationModel applicationModel = ApplicationModel.defaultModel(); ApplicationConfig applicationConfig = new ApplicationConfig("provider-app"); applicationConfig.setExecutorManagementMode(EXECUTOR_MANAGEMENT_MODE_DEFAULT); diff --git a/dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyPortUnificationServer.java b/dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyPortUnificationServer.java index 42d8d89895..f5fdc9a919 100644 --- a/dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyPortUnificationServer.java +++ b/dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyPortUnificationServer.java @@ -178,7 +178,7 @@ public class NettyPortUnificationServer extends AbstractPortUnificationServer { try { if (bootstrap != null) { - long timeout = serverShutdownTimeoutMills; + long timeout = ConfigurationUtils.reCalShutdownTime(serverShutdownTimeoutMills); long quietPeriod = Math.min(2000L, timeout); Future bossGroupShutdownFuture = bossGroup.shutdownGracefully(quietPeriod, timeout, MILLISECONDS); diff --git a/dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyServer.java b/dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyServer.java index 45b3ac90cf..8511370b5d 100644 --- a/dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyServer.java +++ b/dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyServer.java @@ -174,7 +174,7 @@ public class NettyServer extends AbstractServer { } try { if (bootstrap != null) { - long timeout = serverShutdownTimeoutMills; + long timeout = ConfigurationUtils.reCalShutdownTime(serverShutdownTimeoutMills); long quietPeriod = Math.min(2000L, timeout); Future bossGroupShutdownFuture = bossGroup.shutdownGracefully(quietPeriod, timeout, MILLISECONDS); Future workerGroupShutdownFuture = workerGroup.shutdownGracefully(quietPeriod, timeout, MILLISECONDS); diff --git a/dubbo-remoting/dubbo-remoting-netty4/src/test/java/org/apache/dubbo/remoting/transport/netty4/ClientReconnectTest.java b/dubbo-remoting/dubbo-remoting-netty4/src/test/java/org/apache/dubbo/remoting/transport/netty4/ClientReconnectTest.java index 369108ea6b..ff15bda516 100644 --- a/dubbo-remoting/dubbo-remoting-netty4/src/test/java/org/apache/dubbo/remoting/transport/netty4/ClientReconnectTest.java +++ b/dubbo-remoting/dubbo-remoting-netty4/src/test/java/org/apache/dubbo/remoting/transport/netty4/ClientReconnectTest.java @@ -32,11 +32,13 @@ import org.apache.dubbo.remoting.exchange.support.ExchangeHandlerAdapter; import org.apache.dubbo.rpc.model.ApplicationModel; import org.apache.dubbo.rpc.model.FrameworkModel; import org.apache.dubbo.rpc.model.ModuleModel; + import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import static org.apache.dubbo.common.constants.CommonConstants.EXECUTOR_MANAGEMENT_MODE_DEFAULT; +import static org.apache.dubbo.remoting.Constants.LEAST_RECONNECT_DURATION_KEY; /** * Client reconnect test @@ -81,7 +83,8 @@ class ClientReconnectTest { public Client startClient(int port, int heartbeat) throws RemotingException { - URL url = URL.valueOf("exchange://127.0.0.1:" + port + "/client.reconnect.test?client=netty4&check=false&" + Constants.HEARTBEAT_KEY + "=" + heartbeat); + URL url = URL.valueOf("exchange://127.0.0.1:" + port + "/client.reconnect.test?client=netty4&check=false&" + + Constants.HEARTBEAT_KEY + "=" + heartbeat + "&" + LEAST_RECONNECT_DURATION_KEY + "=0"); FrameworkModel frameworkModel = new FrameworkModel(); ApplicationModel applicationModel = frameworkModel.newApplication(); ApplicationConfig applicationConfig = new ApplicationConfig("provider-app"); diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/ReferenceCountInvokerWrapper.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/ReferenceCountInvokerWrapper.java index 1326cf16a4..9cca735949 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/ReferenceCountInvokerWrapper.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/ReferenceCountInvokerWrapper.java @@ -17,6 +17,7 @@ package org.apache.dubbo.rpc.protocol; import org.apache.dubbo.common.URL; +import org.apache.dubbo.common.config.ConfigurationUtils; import org.apache.dubbo.common.constants.LoggerCodeConstants; import org.apache.dubbo.common.logger.ErrorTypeAwareLogger; import org.apache.dubbo.common.logger.LoggerFactory; @@ -25,6 +26,7 @@ import org.apache.dubbo.rpc.Invoker; import org.apache.dubbo.rpc.Result; import org.apache.dubbo.rpc.RpcException; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.locks.ReadWriteLock; import java.util.concurrent.locks.ReentrantReadWriteLock; @@ -53,8 +55,16 @@ public class ReferenceCountInvokerWrapper implements Invoker { @Override public void destroy() { try { - lock.writeLock().lock(); + int timeout = ConfigurationUtils.getServerShutdownTimeout(invoker.getUrl().getScopeModel()); + boolean locked = lock.writeLock().tryLock( + timeout, TimeUnit.MILLISECONDS); + if (!locked) { + logger.warn(LoggerCodeConstants.PROTOCOL_CLOSED_SERVER, "", "", + "Failed to wait for invocation end in " + timeout + "ms."); + } destroyed.set(true); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); } finally { lock.writeLock().unlock(); } diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/proxy/InvocationUtil.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/proxy/InvocationUtil.java index f7b3a7c89b..cc4c258f6a 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/proxy/InvocationUtil.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/proxy/InvocationUtil.java @@ -24,6 +24,7 @@ import org.apache.dubbo.common.profiler.Profiler; import org.apache.dubbo.common.profiler.ProfilerEntry; import org.apache.dubbo.common.profiler.ProfilerSwitch; import org.apache.dubbo.rpc.Invoker; +import org.apache.dubbo.rpc.RpcContext; import org.apache.dubbo.rpc.RpcInvocation; import org.apache.dubbo.rpc.RpcServiceContext; import org.apache.dubbo.rpc.support.RpcUtils; @@ -36,53 +37,59 @@ public class InvocationUtil { private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(InvokerInvocationHandler.class); public static Object invoke(Invoker invoker, RpcInvocation rpcInvocation) throws Throwable { - URL url = invoker.getUrl(); - String serviceKey = url.getServiceKey(); - rpcInvocation.setTargetServiceUniqueName(serviceKey); + RpcContext.RestoreServiceContext originServiceContext = RpcContext.storeServiceContext(); - // invoker.getUrl() returns consumer url. - RpcServiceContext.getServiceContext().setConsumerUrl(url); + try { + URL url = invoker.getUrl(); + String serviceKey = url.getServiceKey(); + rpcInvocation.setTargetServiceUniqueName(serviceKey); - if (ProfilerSwitch.isEnableSimpleProfiler()) { - ProfilerEntry parentProfiler = Profiler.getBizProfiler(); - ProfilerEntry bizProfiler; - if (parentProfiler != null) { - bizProfiler = Profiler.enter(parentProfiler, - "Receive request. Client invoke begin. ServiceKey: " + serviceKey + " MethodName:" + rpcInvocation.getMethodName()); - } else { - bizProfiler = Profiler.start("Receive request. Client invoke begin. ServiceKey: " + serviceKey + " " + "MethodName:" + rpcInvocation.getMethodName()); - } - rpcInvocation.put(Profiler.PROFILER_KEY, bizProfiler); - try { - return invoker.invoke(rpcInvocation).recreate(); - } finally { - Profiler.release(bizProfiler); - Long timeout = RpcUtils.convertToNumber(rpcInvocation.getObjectAttachmentWithoutConvert(TIMEOUT_KEY)); + // invoker.getUrl() returns consumer url. + RpcServiceContext.getServiceContext().setConsumerUrl(url); - if (timeout == null) { - timeout = (long) url.getMethodPositiveParameter(rpcInvocation.getMethodName(), - TIMEOUT_KEY, - DEFAULT_TIMEOUT); + if (ProfilerSwitch.isEnableSimpleProfiler()) { + ProfilerEntry parentProfiler = Profiler.getBizProfiler(); + ProfilerEntry bizProfiler; + if (parentProfiler != null) { + bizProfiler = Profiler.enter(parentProfiler, + "Receive request. Client invoke begin. ServiceKey: " + serviceKey + " MethodName:" + rpcInvocation.getMethodName()); + } else { + bizProfiler = Profiler.start("Receive request. Client invoke begin. ServiceKey: " + serviceKey + " " + "MethodName:" + rpcInvocation.getMethodName()); } - long usage = bizProfiler.getEndTime() - bizProfiler.getStartTime(); - if ((usage / (1000_000L * ProfilerSwitch.getWarnPercent())) > timeout) { - StringBuilder attachment = new StringBuilder(); - rpcInvocation.foreachAttachment((entry) -> { - attachment.append(entry.getKey()).append("=").append(entry.getValue()).append(";\n"); - }); + rpcInvocation.put(Profiler.PROFILER_KEY, bizProfiler); + try { + return invoker.invoke(rpcInvocation).recreate(); + } finally { + Profiler.release(bizProfiler); + Long timeout = RpcUtils.convertToNumber(rpcInvocation.getObjectAttachmentWithoutConvert(TIMEOUT_KEY)); - logger.warn(PROXY_TIMEOUT_REQUEST, "", "", String.format( - "[Dubbo-Consumer] execute service %s#%s cost %d.%06d ms, this invocation almost (maybe already) timeout. Timeout: %dms\n" + "invocation context:\n%s" + "thread info: \n%s", - rpcInvocation.getProtocolServiceKey(), - rpcInvocation.getMethodName(), - usage / 1000_000, - usage % 1000_000, - timeout, - attachment, - Profiler.buildDetail(bizProfiler))); + if (timeout == null) { + timeout = (long) url.getMethodPositiveParameter(rpcInvocation.getMethodName(), + TIMEOUT_KEY, + DEFAULT_TIMEOUT); + } + long usage = bizProfiler.getEndTime() - bizProfiler.getStartTime(); + if ((usage / (1000_000L * ProfilerSwitch.getWarnPercent())) > timeout) { + StringBuilder attachment = new StringBuilder(); + rpcInvocation.foreachAttachment((entry) -> { + attachment.append(entry.getKey()).append("=").append(entry.getValue()).append(";\n"); + }); + + logger.warn(PROXY_TIMEOUT_REQUEST, "", "", String.format( + "[Dubbo-Consumer] execute service %s#%s cost %d.%06d ms, this invocation almost (maybe already) timeout. Timeout: %dms\n" + "invocation context:\n%s" + "thread info: \n%s", + rpcInvocation.getProtocolServiceKey(), + rpcInvocation.getMethodName(), + usage / 1000_000, + usage % 1000_000, + timeout, + attachment, + Profiler.buildDetail(bizProfiler))); + } } } + return invoker.invoke(rpcInvocation).recreate(); + } finally { + RpcContext.restoreServiceContext(originServiceContext); } - return invoker.invoke(rpcInvocation).recreate(); } } diff --git a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/ByteAccessor.java b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/ByteAccessor.java index e0913486da..405edadaa0 100644 --- a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/ByteAccessor.java +++ b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/ByteAccessor.java @@ -20,6 +20,8 @@ package org.apache.dubbo.rpc.protocol.dubbo; import org.apache.dubbo.common.extension.SPI; import org.apache.dubbo.remoting.Channel; import org.apache.dubbo.remoting.exchange.Request; +import org.apache.dubbo.remoting.exchange.Response; +import org.apache.dubbo.rpc.Invocation; import java.io.InputStream; @@ -38,4 +40,10 @@ public interface ByteAccessor { * The parameters are the same as {@link DecodeableRpcInvocation} */ DecodeableRpcInvocation getRpcInvocation(Channel channel, Request req, InputStream is, byte proto); + + /** + * Get an enhanced DecodeableRpcResult subclass to allow custom decode. + * The parameters are the same as {@link DecodeableRpcResult} + */ + DecodeableRpcResult getRpcResult(Channel channel, Response res, InputStream is, Invocation invocation, byte proto); } diff --git a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboCodec.java b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboCodec.java index a61aec3934..c775ae50a0 100644 --- a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboCodec.java +++ b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboCodec.java @@ -114,14 +114,29 @@ public class DubboCodec extends ExchangeCodec { } } else { DecodeableRpcResult result; + Invocation inv = (Invocation) getRequestData(channel, res, id); if (channel.getUrl().getParameter(DECODE_IN_IO_THREAD_KEY, DEFAULT_DECODE_IN_IO_THREAD)) { - result = new DecodeableRpcResult(channel, res, is, - (Invocation) getRequestData(channel, res, id), proto); + if (customByteAccessor != null) { + result = customByteAccessor.getRpcResult(channel, res, + new UnsafeByteArrayInputStream(readMessageData(is)), + inv, proto); + } else { + result = new DecodeableRpcResult(channel, res, + new UnsafeByteArrayInputStream(readMessageData(is)), + inv, proto); + } result.decode(); } else { - result = new DecodeableRpcResult(channel, res, - new UnsafeByteArrayInputStream(readMessageData(is)), - (Invocation) getRequestData(channel, res, id), proto); + if (customByteAccessor != null) { + result = customByteAccessor.getRpcResult(channel, res, + new UnsafeByteArrayInputStream(readMessageData(is)), + inv, proto); + } else { + result = new DecodeableRpcResult(channel, res, + new UnsafeByteArrayInputStream(readMessageData(is)), + inv, proto); + } + } data = result; } diff --git a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboInvoker.java b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboInvoker.java index 06466c6938..25657a26cb 100644 --- a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboInvoker.java +++ b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboInvoker.java @@ -34,7 +34,6 @@ import org.apache.dubbo.rpc.Invocation; import org.apache.dubbo.rpc.InvokeMode; import org.apache.dubbo.rpc.Invoker; import org.apache.dubbo.rpc.Result; -import org.apache.dubbo.rpc.RpcContext; import org.apache.dubbo.rpc.RpcException; import org.apache.dubbo.rpc.RpcInvocation; import org.apache.dubbo.rpc.protocol.AbstractInvoker; @@ -111,8 +110,6 @@ public class DubboInvoker extends AbstractInvoker { invocation.setAttachment(TIMEOUT_KEY, String.valueOf(timeout)); - RpcContext.getServiceContext().setRemoteAddress(currentClient.getRemoteAddress()); - Integer payload = getUrl().getParameter(PAYLOAD, Integer.class); Request request = new Request(); @@ -182,7 +179,7 @@ public class DubboInvoker extends AbstractInvoker { if (invokers != null) { invokers.remove(this); } - clientsProvider.close(serverShutdownTimeout); + clientsProvider.close(ConfigurationUtils.reCalShutdownTime(serverShutdownTimeout)); } finally { destroyLock.unlock(); } 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 2483f78d00..b9baa3b386 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 @@ -554,7 +554,7 @@ public class DubboProtocol extends AbstractProtocol { logger.info("Closing dubbo server: " + server.getLocalAddress()); } - server.close(getServerShutdownTimeout(protocolServer)); + server.close(ConfigurationUtils.reCalShutdownTime(getServerShutdownTimeout(protocolServer))); } catch (Throwable t) { logger.warn(PROTOCOL_ERROR_CLOSE_SERVER, "", "", "Close dubbo server [" + server.getLocalAddress() + "] failed: " + t.getMessage(), t); diff --git a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/SharedClientsProvider.java b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/SharedClientsProvider.java index 1f527f7506..83ca0cfce2 100644 --- a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/SharedClientsProvider.java +++ b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/SharedClientsProvider.java @@ -16,6 +16,7 @@ */ package org.apache.dubbo.rpc.protocol.dubbo; +import org.apache.dubbo.common.config.ConfigurationUtils; import org.apache.dubbo.common.logger.ErrorTypeAwareLogger; import org.apache.dubbo.common.logger.LoggerFactory; import org.apache.dubbo.common.utils.CollectionUtils; @@ -116,7 +117,7 @@ public class SharedClientsProvider implements ClientsProvider { logger.info("Close dubbo connect: " + client.getLocalAddress() + "-->" + client.getRemoteAddress()); } - client.close(client.getShutdownWaitTime()); + client.close(ConfigurationUtils.reCalShutdownTime(client.getShutdownWaitTime())); // TODO /*