From 7facd5be19ea80ffee05172931b589ddc90ace8a Mon Sep 17 00:00:00 2001 From: Albumen Kevin Date: Tue, 11 Apr 2023 19:00:02 +0800 Subject: [PATCH] Fix memory leak when hashed wheel time re-put (#12022) --- .../cluster/directory/StaticDirectory.java | 6 ++- .../directory/StaticDirectoryTest.java | 10 ++-- .../integration/DynamicDirectory.java | 10 +++- .../support/header/AbstractTimerTask.java | 44 ++++++++++------- .../support/header/CloseTimerTask.java | 5 +- .../support/header/HeaderExchangeClient.java | 47 +++++++++++-------- .../support/header/HeaderExchangeServer.java | 12 ++--- .../support/header/HeartbeatTimerTask.java | 5 +- .../support/header/ReconnectTimerTask.java | 7 +-- .../support/header/CloseTimerTaskTest.java | 8 +++- .../support/header/HeartBeatTaskTest.java | 12 +++-- .../header/ReconnectTimerTaskTest.java | 12 +++-- 12 files changed, 105 insertions(+), 73 deletions(-) diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/directory/StaticDirectory.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/directory/StaticDirectory.java index d8a022f3d2..8a17dc0fc0 100644 --- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/directory/StaticDirectory.java +++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/directory/StaticDirectory.java @@ -16,8 +16,6 @@ */ package org.apache.dubbo.rpc.cluster.directory; -import java.util.List; - import org.apache.dubbo.common.URL; import org.apache.dubbo.common.logger.ErrorTypeAwareLogger; import org.apache.dubbo.common.logger.LoggerFactory; @@ -29,6 +27,8 @@ import org.apache.dubbo.rpc.cluster.RouterChain; import org.apache.dubbo.rpc.cluster.SingleRouterChain; import org.apache.dubbo.rpc.cluster.router.state.BitList; +import java.util.List; + import static org.apache.dubbo.common.constants.LoggerCodeConstants.CLUSTER_FAILED_SITE_SELECTION; /** @@ -75,6 +75,8 @@ public class StaticDirectory extends AbstractDirectory { for (Invoker invoker : getValidInvokers()) { if (invoker.isAvailable()) { return true; + } else { + addInvalidateInvoker(invoker); } } return false; diff --git a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/directory/StaticDirectoryTest.java b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/directory/StaticDirectoryTest.java index d10fd8649e..68ee46afb1 100644 --- a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/directory/StaticDirectoryTest.java +++ b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/directory/StaticDirectoryTest.java @@ -26,8 +26,8 @@ import org.apache.dubbo.rpc.cluster.router.MockInvoker; import org.apache.dubbo.rpc.cluster.router.condition.ConditionStateRouterFactory; import org.apache.dubbo.rpc.cluster.router.state.BitList; import org.apache.dubbo.rpc.cluster.router.state.StateRouter; - import org.apache.dubbo.rpc.model.ApplicationModel; + import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; @@ -52,9 +52,9 @@ class StaticDirectoryTest { List routers = new ArrayList(); routers.add(router); List> originInvokers = new ArrayList>(); - Invoker invoker1 = new MockInvoker(URL.valueOf("dubbo://10.20.3.3:20880/com.foo.BarService")); - Invoker invoker2 = new MockInvoker(URL.valueOf("dubbo://" + NetUtils.getLocalHost() + ":20880/com.foo.BarService")); - Invoker invoker3 = new MockInvoker(URL.valueOf("dubbo://" + NetUtils.getLocalHost() + ":20880/com.foo.BarService")); + Invoker invoker1 = new MockInvoker(URL.valueOf("dubbo://10.20.3.3:20880/com.foo.BarService"), true); + Invoker invoker2 = new MockInvoker(URL.valueOf("dubbo://" + NetUtils.getLocalHost() + ":20880/com.foo.BarService"), true); + Invoker invoker3 = new MockInvoker(URL.valueOf("dubbo://" + NetUtils.getLocalHost() + ":20880/com.foo.BarService"), true); originInvokers.add(invoker1); originInvokers.add(invoker2); originInvokers.add(invoker3); @@ -65,7 +65,7 @@ class StaticDirectoryTest { ApplicationModel.defaultModel().getBeanFactory().registerBean(MetricsDispatcher.class); StaticDirectory staticDirectory = new StaticDirectory<>(filteredInvokers); boolean isAvailable = staticDirectory.isAvailable(); - Assertions.assertTrue(!isAvailable); + Assertions.assertTrue(isAvailable); List> newInvokers = staticDirectory.list(new MockDirInvocation()); Assertions.assertTrue(newInvokers.size() > 0); staticDirectory.destroy(); diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/DynamicDirectory.java b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/DynamicDirectory.java index c171b9b5b0..7ee4da0ed8 100644 --- a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/DynamicDirectory.java +++ b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/DynamicDirectory.java @@ -287,8 +287,14 @@ public abstract class DynamicDirectory extends AbstractDirectory implement if (isDestroyed() || this.forbidden) { return false; } - return CollectionUtils.isNotEmpty(getValidInvokers()) - && getValidInvokers().stream().anyMatch(Invoker::isAvailable); + for (Invoker validInvoker : getValidInvokers()) { + if (validInvoker.isAvailable()) { + return true; + } else { + addInvalidateInvoker(validInvoker); + } + } + return false; } @Override diff --git a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/AbstractTimerTask.java b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/AbstractTimerTask.java index 3779d9917f..ce5c5e6014 100644 --- a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/AbstractTimerTask.java +++ b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/AbstractTimerTask.java @@ -17,8 +17,8 @@ package org.apache.dubbo.remoting.exchange.support.header; +import org.apache.dubbo.common.timer.HashedWheelTimer; import org.apache.dubbo.common.timer.Timeout; -import org.apache.dubbo.common.timer.Timer; import org.apache.dubbo.common.timer.TimerTask; import org.apache.dubbo.remoting.Channel; @@ -31,17 +31,22 @@ import java.util.concurrent.TimeUnit; public abstract class AbstractTimerTask implements TimerTask { private final ChannelProvider channelProvider; + private final HashedWheelTimer hashedWheelTimer; private final Long tick; protected volatile boolean cancel = false; - AbstractTimerTask(ChannelProvider channelProvider, Long tick) { - if (channelProvider == null || tick == null) { + private volatile Timeout timeout; + + AbstractTimerTask(ChannelProvider channelProvider, HashedWheelTimer hashedWheelTimer, Long tick) { + if (channelProvider == null || hashedWheelTimer == null || tick == null) { throw new IllegalArgumentException(); } - this.tick = tick; this.channelProvider = channelProvider; + this.hashedWheelTimer = hashedWheelTimer; + this.tick = tick; + start(); } static Long lastRead(Channel channel) { @@ -56,12 +61,17 @@ public abstract class AbstractTimerTask implements TimerTask { return System.currentTimeMillis(); } - public void cancel() { - this.cancel = true; + private void start() { + this.timeout = hashedWheelTimer.newTimeout(this, tick, TimeUnit.MILLISECONDS); } - private void reput(Timeout timeout, Long tick) { - if (timeout == null || tick == null) { + public synchronized void cancel() { + this.cancel = true; + this.timeout.cancel(); + } + + private synchronized void reput(Timeout timeout) { + if (timeout == null) { throw new IllegalArgumentException(); } @@ -69,24 +79,22 @@ public abstract class AbstractTimerTask implements TimerTask { return; } - Timer timer = timeout.timer(); - if (timer.isStop() || timeout.isCancelled()) { + if (hashedWheelTimer.isStop() || timeout.isCancelled()) { return; } - timer.newTimeout(timeout.task(), tick, TimeUnit.MILLISECONDS); + this.timeout = hashedWheelTimer.newTimeout(timeout.task(), tick, TimeUnit.MILLISECONDS); } @Override - public void run(Timeout timeout) throws Exception { - Collection c = channelProvider.getChannels(); - for (Channel channel : c) { - if (channel.isClosed()) { - continue; + public synchronized void run(Timeout timeout) throws Exception { + Collection channels = channelProvider.getChannels(); + for (Channel channel : channels) { + if (!channel.isClosed()) { + doTask(channel); } - doTask(channel); } - reput(timeout, tick); + reput(timeout); } protected abstract void doTask(Channel channel); diff --git a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/CloseTimerTask.java b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/CloseTimerTask.java index a1c850f619..7c2eea2289 100644 --- a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/CloseTimerTask.java +++ b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/CloseTimerTask.java @@ -19,6 +19,7 @@ package org.apache.dubbo.remoting.exchange.support.header; import org.apache.dubbo.common.logger.ErrorTypeAwareLogger; import org.apache.dubbo.common.logger.LoggerFactory; +import org.apache.dubbo.common.timer.HashedWheelTimer; import org.apache.dubbo.remoting.Channel; import static org.apache.dubbo.common.constants.LoggerCodeConstants.PROTOCOL_FAILED_RESPONSE; @@ -33,8 +34,8 @@ public class CloseTimerTask extends AbstractTimerTask { private final int idleTimeout; - public CloseTimerTask(ChannelProvider channelProvider, Long heartbeatTimeoutTick, int idleTimeout) { - super(channelProvider, heartbeatTimeoutTick); + public CloseTimerTask(ChannelProvider channelProvider, HashedWheelTimer hashedWheelTimer, Long heartbeatTimeoutTick, int idleTimeout) { + super(channelProvider, hashedWheelTimer, heartbeatTimeoutTick); this.idleTimeout = idleTimeout; } 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 b25951a98c..f776e9e5d7 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 @@ -19,7 +19,6 @@ package org.apache.dubbo.remoting.exchange.support.header; import org.apache.dubbo.common.URL; import org.apache.dubbo.common.resource.GlobalResourceInitializer; import org.apache.dubbo.common.timer.HashedWheelTimer; -import org.apache.dubbo.common.timer.Timeout; import org.apache.dubbo.common.utils.Assert; import org.apache.dubbo.common.utils.NamedThreadFactory; import org.apache.dubbo.remoting.ChannelHandler; @@ -51,12 +50,13 @@ public class HeaderExchangeClient implements ExchangeClient { private final ExchangeChannel channel; public static GlobalResourceInitializer IDLE_CHECK_TIMER = new GlobalResourceInitializer<>(() -> - new HashedWheelTimer(new NamedThreadFactory("dubbo-client-idleCheck", true), 1, + new HashedWheelTimer(new NamedThreadFactory("dubbo-client-heartbeat-reconnect", true), 1, TimeUnit.SECONDS, TICKS_PER_WHEEL), HashedWheelTimer::stop); - private Timeout reconnectTimer; - private Timeout heartBeatTimer; + private ReconnectTimerTask reconnectTimerTask; + private HeartbeatTimerTask heartBeatTimerTask; + private final int idleTimeout; public HeaderExchangeClient(Client client, boolean startTimer) { Assert.notNull(client, "Client can't be null"); @@ -65,8 +65,11 @@ public class HeaderExchangeClient implements ExchangeClient { if (startTimer) { URL url = client.getUrl(); + idleTimeout = getIdleTimeout(url); startReconnectTask(url); startHeartBeatTask(url); + } else { + idleTimeout = 0; } } @@ -107,7 +110,16 @@ public class HeaderExchangeClient implements ExchangeClient { @Override public boolean isConnected() { - return channel.isConnected(); + if (channel.isConnected()) { + if (idleTimeout <= 0) { + return true; + } + Long lastRead = (Long) channel.getAttribute(HeartbeatHandler.KEY_READ_TIMESTAMP); + Long now = System.currentTimeMillis(); + + return lastRead == null || now - lastRead < idleTimeout; + } + return false; } @Override @@ -136,7 +148,7 @@ public class HeaderExchangeClient implements ExchangeClient { } @Override - public void close() { + public synchronized void close() { doClose(); channel.close(); } @@ -167,7 +179,7 @@ public class HeaderExchangeClient implements ExchangeClient { } @Override - public void reconnect() throws RemotingException { + public synchronized void reconnect() throws RemotingException { client.reconnect(); } @@ -193,32 +205,27 @@ public class HeaderExchangeClient implements ExchangeClient { private void startHeartBeatTask(URL url) { if (!client.canHandleIdle()) { - AbstractTimerTask.ChannelProvider cp = () -> Collections.singletonList(HeaderExchangeClient.this); int heartbeat = getHeartbeat(url); long heartbeatTick = calculateLeastDuration(heartbeat); - HeartbeatTimerTask heartBeatTimerTask = new HeartbeatTimerTask(cp, heartbeatTick, heartbeat); - heartBeatTimer = IDLE_CHECK_TIMER.get().newTimeout(heartBeatTimerTask, heartbeatTick, TimeUnit.MILLISECONDS); + heartBeatTimerTask = new HeartbeatTimerTask(() -> Collections.singleton(this), IDLE_CHECK_TIMER.get(), heartbeatTick, heartbeat); } } private void startReconnectTask(URL url) { if (shouldReconnect(url)) { - AbstractTimerTask.ChannelProvider cp = () -> Collections.singletonList(HeaderExchangeClient.this); - int idleTimeout = getIdleTimeout(url); long heartbeatTimeoutTick = calculateLeastDuration(idleTimeout); - ReconnectTimerTask reconnectTimerTask = new ReconnectTimerTask(cp, heartbeatTimeoutTick, idleTimeout); - reconnectTimer = IDLE_CHECK_TIMER.get().newTimeout(reconnectTimerTask, heartbeatTimeoutTick, TimeUnit.MILLISECONDS); + reconnectTimerTask = new ReconnectTimerTask(() -> Collections.singleton(this), IDLE_CHECK_TIMER.get(), heartbeatTimeoutTick, idleTimeout); } } private void doClose() { - if (heartBeatTimer != null) { - heartBeatTimer.cancel(); - heartBeatTimer = null; + if (heartBeatTimerTask != null) { + heartBeatTimerTask.cancel(); + heartBeatTimerTask = null; } - if (reconnectTimer != null) { - reconnectTimer.cancel(); - reconnectTimer = null; + if (reconnectTimerTask != null) { + reconnectTimerTask.cancel(); + reconnectTimerTask = null; } } diff --git a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/HeaderExchangeServer.java b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/HeaderExchangeServer.java index 25d2370d4f..f27aae80e8 100644 --- a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/HeaderExchangeServer.java +++ b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/HeaderExchangeServer.java @@ -22,7 +22,6 @@ import org.apache.dubbo.common.logger.ErrorTypeAwareLogger; import org.apache.dubbo.common.logger.LoggerFactory; import org.apache.dubbo.common.resource.GlobalResourceInitializer; import org.apache.dubbo.common.timer.HashedWheelTimer; -import org.apache.dubbo.common.timer.Timeout; import org.apache.dubbo.common.utils.Assert; import org.apache.dubbo.common.utils.CollectionUtils; import org.apache.dubbo.common.utils.NamedThreadFactory; @@ -44,9 +43,9 @@ import java.util.concurrent.atomic.AtomicBoolean; import static java.util.Collections.unmodifiableCollection; import static org.apache.dubbo.common.constants.CommonConstants.READONLY_EVENT; +import static org.apache.dubbo.common.constants.LoggerCodeConstants.INTERNAL_ERROR; import static org.apache.dubbo.common.constants.LoggerCodeConstants.TRANSPORT_FAILED_CLOSE; import static org.apache.dubbo.common.constants.LoggerCodeConstants.TRANSPORT_FAILED_RESPONSE; -import static org.apache.dubbo.common.constants.LoggerCodeConstants.INTERNAL_ERROR; 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.TICKS_PER_WHEEL; @@ -66,7 +65,7 @@ public class HeaderExchangeServer implements ExchangeServer { public static GlobalResourceInitializer IDLE_CHECK_TIMER = new GlobalResourceInitializer<>(() -> new HashedWheelTimer(new NamedThreadFactory("dubbo-server-idleCheck", true), 1, TimeUnit.SECONDS, TICKS_PER_WHEEL), HashedWheelTimer::stop); - private Timeout closeTimer; + private CloseTimerTask closeTimer; public HeaderExchangeServer(RemotingServer server) { Assert.notNull(server, "server == null"); @@ -141,7 +140,7 @@ public class HeaderExchangeServer implements ExchangeServer { } } catch (RemotingException e) { if (closed.get() && e.getCause() instanceof ClosedChannelException) { - // ignore ClosedChannelException which means the connection has been closed. + // ignore ClosedChannelException which means the connection has been closed. continue; } logger.warn(TRANSPORT_FAILED_RESPONSE, "", "", "send cannot write message error.", e); @@ -265,10 +264,7 @@ public class HeaderExchangeServer implements ExchangeServer { AbstractTimerTask.ChannelProvider cp = () -> unmodifiableCollection(HeaderExchangeServer.this.getChannels()); int idleTimeout = getIdleTimeout(url); long idleTimeoutTick = calculateLeastDuration(idleTimeout); - CloseTimerTask closeTimerTask = new CloseTimerTask(cp, idleTimeoutTick, idleTimeout); - - // init task and start timer. - this.closeTimer = IDLE_CHECK_TIMER.get().newTimeout(closeTimerTask, idleTimeoutTick, TimeUnit.MILLISECONDS); + this.closeTimer = new CloseTimerTask(cp, IDLE_CHECK_TIMER.get(), idleTimeoutTick, idleTimeout); } } } diff --git a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/HeartbeatTimerTask.java b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/HeartbeatTimerTask.java index 79e2efe876..e418a1de99 100644 --- a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/HeartbeatTimerTask.java +++ b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/HeartbeatTimerTask.java @@ -20,6 +20,7 @@ package org.apache.dubbo.remoting.exchange.support.header; import org.apache.dubbo.common.Version; import org.apache.dubbo.common.logger.ErrorTypeAwareLogger; import org.apache.dubbo.common.logger.LoggerFactory; +import org.apache.dubbo.common.timer.HashedWheelTimer; import org.apache.dubbo.remoting.Channel; import org.apache.dubbo.remoting.exchange.Request; @@ -35,8 +36,8 @@ public class HeartbeatTimerTask extends AbstractTimerTask { private final int heartbeat; - HeartbeatTimerTask(ChannelProvider channelProvider, Long heartbeatTick, int heartbeat) { - super(channelProvider, heartbeatTick); + HeartbeatTimerTask(ChannelProvider channelProvider, HashedWheelTimer hashedWheelTimer, Long heartbeatTick, int heartbeat) { + super(channelProvider, hashedWheelTimer, heartbeatTick); this.heartbeat = heartbeat; } diff --git a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/ReconnectTimerTask.java b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/ReconnectTimerTask.java index 54a035b764..506ec003c1 100644 --- a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/ReconnectTimerTask.java +++ b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/ReconnectTimerTask.java @@ -19,11 +19,12 @@ package org.apache.dubbo.remoting.exchange.support.header; import org.apache.dubbo.common.logger.ErrorTypeAwareLogger; import org.apache.dubbo.common.logger.LoggerFactory; +import org.apache.dubbo.common.timer.HashedWheelTimer; import org.apache.dubbo.remoting.Channel; import org.apache.dubbo.remoting.Client; -import static org.apache.dubbo.common.constants.LoggerCodeConstants.TRANSPORT_FAILED_RECONNECT; import static org.apache.dubbo.common.constants.LoggerCodeConstants.INTERNAL_ERROR; +import static org.apache.dubbo.common.constants.LoggerCodeConstants.TRANSPORT_FAILED_RECONNECT; /** * ReconnectTimerTask @@ -34,8 +35,8 @@ public class ReconnectTimerTask extends AbstractTimerTask { private final int idleTimeout; - public ReconnectTimerTask(ChannelProvider channelProvider, Long heartbeatTimeoutTick, int idleTimeout) { - super(channelProvider, heartbeatTimeoutTick); + public ReconnectTimerTask(ChannelProvider channelProvider, HashedWheelTimer hashedWheelTimer, Long heartbeatTimeoutTick, int idleTimeout) { + super(channelProvider, hashedWheelTimer, heartbeatTimeoutTick); this.idleTimeout = idleTimeout; } diff --git a/dubbo-remoting/dubbo-remoting-api/src/test/java/org/apache/dubbo/remoting/exchange/support/header/CloseTimerTaskTest.java b/dubbo-remoting/dubbo-remoting-api/src/test/java/org/apache/dubbo/remoting/exchange/support/header/CloseTimerTaskTest.java index f9d232575d..882a481c24 100644 --- a/dubbo-remoting/dubbo-remoting-api/src/test/java/org/apache/dubbo/remoting/exchange/support/header/CloseTimerTaskTest.java +++ b/dubbo-remoting/dubbo-remoting-api/src/test/java/org/apache/dubbo/remoting/exchange/support/header/CloseTimerTaskTest.java @@ -20,6 +20,7 @@ import org.apache.dubbo.common.URL; import org.apache.dubbo.common.timer.HashedWheelTimer; import org.apache.dubbo.remoting.Channel; +import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -54,7 +55,12 @@ class CloseTimerTaskTest { }; AbstractTimerTask.ChannelProvider cp = () -> Collections.singletonList(channel); - closeTimerTask = new CloseTimerTask(cp, tickDuration / HEARTBEAT_CHECK_TICK, (int) tickDuration); + closeTimerTask = new CloseTimerTask(cp, closeTimer, tickDuration / HEARTBEAT_CHECK_TICK, (int) tickDuration); + } + + @AfterEach + public void teardown() { + closeTimerTask.cancel(); } @Test diff --git a/dubbo-remoting/dubbo-remoting-api/src/test/java/org/apache/dubbo/remoting/exchange/support/header/HeartBeatTaskTest.java b/dubbo-remoting/dubbo-remoting-api/src/test/java/org/apache/dubbo/remoting/exchange/support/header/HeartBeatTaskTest.java index cb07c11df7..07a0f63a0a 100644 --- a/dubbo-remoting/dubbo-remoting-api/src/test/java/org/apache/dubbo/remoting/exchange/support/header/HeartBeatTaskTest.java +++ b/dubbo-remoting/dubbo-remoting-api/src/test/java/org/apache/dubbo/remoting/exchange/support/header/HeartBeatTaskTest.java @@ -19,9 +19,9 @@ package org.apache.dubbo.remoting.exchange.support.header; import org.apache.dubbo.common.URL; import org.apache.dubbo.common.timer.HashedWheelTimer; -import org.apache.dubbo.remoting.Channel; import org.apache.dubbo.remoting.exchange.Request; +import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -55,8 +55,12 @@ class HeartBeatTaskTest { } }; - AbstractTimerTask.ChannelProvider cp = () -> Collections.singletonList(channel); - heartbeatTimerTask = new HeartbeatTimerTask(cp, tickDuration / HEARTBEAT_CHECK_TICK, (int) tickDuration); + heartbeatTimerTask = new HeartbeatTimerTask(() -> Collections.singleton(channel), heartbeatTimer, tickDuration / HEARTBEAT_CHECK_TICK, (int) tickDuration); + } + + @AfterEach + public void teardown() { + heartbeatTimerTask.cancel(); } @Test @@ -67,8 +71,6 @@ class HeartBeatTaskTest { channel.setAttribute(HeartbeatHandler.KEY_READ_TIMESTAMP, now); channel.setAttribute(HeartbeatHandler.KEY_WRITE_TIMESTAMP, now); - heartbeatTimer.newTimeout(heartbeatTimerTask, 250, TimeUnit.MILLISECONDS); - Thread.sleep(2000L); List objects = channel.getSentObjects(); Assertions.assertTrue(objects.size() > 0); diff --git a/dubbo-remoting/dubbo-remoting-api/src/test/java/org/apache/dubbo/remoting/exchange/support/header/ReconnectTimerTaskTest.java b/dubbo-remoting/dubbo-remoting-api/src/test/java/org/apache/dubbo/remoting/exchange/support/header/ReconnectTimerTaskTest.java index 5b8f30006d..e68b26355e 100644 --- a/dubbo-remoting/dubbo-remoting-api/src/test/java/org/apache/dubbo/remoting/exchange/support/header/ReconnectTimerTaskTest.java +++ b/dubbo-remoting/dubbo-remoting-api/src/test/java/org/apache/dubbo/remoting/exchange/support/header/ReconnectTimerTaskTest.java @@ -18,8 +18,8 @@ package org.apache.dubbo.remoting.exchange.support.header; import org.apache.dubbo.common.URL; import org.apache.dubbo.common.timer.HashedWheelTimer; -import org.apache.dubbo.remoting.Channel; +import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -56,8 +56,12 @@ class ReconnectTimerTaskTest { } }; - AbstractTimerTask.ChannelProvider cp = () -> Collections.singletonList(channel); - reconnectTimerTask = new ReconnectTimerTask(cp, tickDuration / HEARTBEAT_CHECK_TICK, (int) tickDuration); + reconnectTimerTask = new ReconnectTimerTask(() -> Collections.singleton(channel), reconnectTimer, tickDuration / HEARTBEAT_CHECK_TICK, (int) tickDuration); + } + + @AfterEach + public void teardown() { + reconnectTimerTask.cancel(); } @Test @@ -68,8 +72,6 @@ class ReconnectTimerTaskTest { channel.setAttribute(HeartbeatHandler.KEY_READ_TIMESTAMP, now - 1000); channel.setAttribute(HeartbeatHandler.KEY_WRITE_TIMESTAMP, now - 1000); - reconnectTimer.newTimeout(reconnectTimerTask, 250, TimeUnit.MILLISECONDS); - Thread.sleep(2000L); Assertions.assertTrue(channel.getReconnectCount() > 0); isConnected = true;