diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/Constants.java b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/Constants.java index f72b23e16721..46b26d6427fe 100644 --- a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/Constants.java +++ b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/Constants.java @@ -88,6 +88,18 @@ public interface Constants { String SESSION_TIMEOUT_KEY = "session"; + /** + * Timeout in milliseconds for the liveness probe the multicast registry runs while cleaning + * expired providers. Kept separate from {@code timeout} so that bounding the probe does not + * change how long the registry waits on its own operations. + */ + String CLEAN_CHECK_TIMEOUT_KEY = "clean.check.timeout"; + + /** + * Default liveness probe timeout: 1000 (ms) + */ + int DEFAULT_CLEAN_CHECK_TIMEOUT = 1000; + /** * To decide the frequency of checking Distributed Service Discovery Registry callback hook (in ms) */ diff --git a/dubbo-registry/dubbo-registry-multicast/src/main/java/org/apache/dubbo/registry/multicast/MulticastRegistry.java b/dubbo-registry/dubbo-registry-multicast/src/main/java/org/apache/dubbo/registry/multicast/MulticastRegistry.java index 83e3a0605552..09f21239fc32 100644 --- a/dubbo-registry/dubbo-registry-multicast/src/main/java/org/apache/dubbo/registry/multicast/MulticastRegistry.java +++ b/dubbo-registry/dubbo-registry-multicast/src/main/java/org/apache/dubbo/registry/multicast/MulticastRegistry.java @@ -59,7 +59,9 @@ import static org.apache.dubbo.common.constants.RegistryConstants.EMPTY_PROTOCOL; import static org.apache.dubbo.common.constants.RegistryConstants.OVERRIDE_PROTOCOL; import static org.apache.dubbo.common.constants.RegistryConstants.ROUTE_PROTOCOL; +import static org.apache.dubbo.registry.Constants.CLEAN_CHECK_TIMEOUT_KEY; import static org.apache.dubbo.registry.Constants.CONSUMER_PROTOCOL; +import static org.apache.dubbo.registry.Constants.DEFAULT_CLEAN_CHECK_TIMEOUT; import static org.apache.dubbo.registry.Constants.DEFAULT_SESSION_TIMEOUT; import static org.apache.dubbo.registry.Constants.REGISTER; import static org.apache.dubbo.registry.Constants.REGISTER_KEY; @@ -94,7 +96,8 @@ public class MulticastRegistry extends FailbackRegistry { * Timeout in milliseconds for the liveness probe performed while cleaning expired providers. * A provider host that is unreachable (for example a host that dropped off the network without * closing its TCP stack) would otherwise block the clean thread until the OS-level TCP timeout, - * which is minutes and stalls expiry detection for every other provider. + * which is minutes and stalls expiry detection for every other provider. A provider that is + * merely slow to accept a connection within this budget is treated as expired. */ private final int cleanCheckTimeout; @@ -143,7 +146,7 @@ public MulticastRegistry(URL url, ApplicationModel applicationModel) { throw new IllegalStateException(e.getMessage(), e); } this.cleanPeriod = url.getParameter(SESSION_TIMEOUT_KEY, DEFAULT_SESSION_TIMEOUT); - this.cleanCheckTimeout = url.getParameter(TIMEOUT_KEY, DEFAULT_TIMEOUT); + this.cleanCheckTimeout = url.getParameter(CLEAN_CHECK_TIMEOUT_KEY, DEFAULT_CLEAN_CHECK_TIMEOUT); if (url.getParameter("clean", true)) { this.cleanFuture = cleanExecutor.scheduleWithFixedDelay( () -> { diff --git a/dubbo-registry/dubbo-registry-multicast/src/test/java/org/apache/dubbo/registry/multicast/MulticastRegistryTest.java b/dubbo-registry/dubbo-registry-multicast/src/test/java/org/apache/dubbo/registry/multicast/MulticastRegistryTest.java index 5bc8d9ce5e3b..a3cea31853b2 100644 --- a/dubbo-registry/dubbo-registry-multicast/src/test/java/org/apache/dubbo/registry/multicast/MulticastRegistryTest.java +++ b/dubbo-registry/dubbo-registry-multicast/src/test/java/org/apache/dubbo/registry/multicast/MulticastRegistryTest.java @@ -18,10 +18,14 @@ import org.apache.dubbo.common.URL; import org.apache.dubbo.common.utils.NetUtils; +import org.apache.dubbo.registry.Constants; import org.apache.dubbo.registry.NotifyListener; +import java.net.ConnectException; import java.net.InetAddress; +import java.net.InetSocketAddress; import java.net.MulticastSocket; +import java.net.Socket; import java.net.UnknownHostException; import java.util.List; import java.util.Map; @@ -31,6 +35,8 @@ import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.mockito.MockedConstruction; +import org.mockito.Mockito; import static org.apache.dubbo.common.constants.RegistryConstants.EMPTY_PROTOCOL; import static org.hamcrest.CoreMatchers.is; @@ -271,20 +277,44 @@ void testMulticastAddress() { } /** - * An unreachable provider must not block the clean thread. TEST-NET-1 (RFC 5737) is reserved and - * unroutable, so without a bounded connect timeout this call would wait for the OS-level TCP - * timeout (tens of seconds) before returning. + * The liveness probe must pass an explicit timeout to {@code Socket.connect}. + * + *

This asserts the argument rather than the elapsed time on purpose. A wall-clock assertion + * against an unroutable address still passes without the timeout wherever the connect fails + * immediately (no route to host, or a firewall that rejects instead of dropping), so it does + * not prove the timeout is applied. Capturing the call is platform independent. */ @Test - void testIsExpiredBoundedForUnreachableProvider() { - URL unreachable = URL.valueOf("dubbo://192.0.2.1:20880/" + service + "?timeout=300"); - - long start = System.nanoTime(); - boolean expired = registry.isExpired(unreachable); - long elapsedMillis = (System.nanoTime() - start) / 1_000_000L; + void testLivenessProbePassesTheConfiguredTimeoutToConnect() throws Exception { + // The option belongs to the registry url: the probe budget is a property of the registry + // doing the cleaning, not of the provider being checked. + MulticastRegistry timedRegistry = + new MulticastRegistry(URL.valueOf("multicast://239.239.239.240/?clean.check.timeout=250")); + try (MockedConstruction mocked = Mockito.mockConstruction(Socket.class)) { + timedRegistry.isExpired(URL.valueOf("dubbo://127.0.0.1:20880/" + service)); + + assertFalse(mocked.constructed().isEmpty(), "the probe must attempt a connect"); + for (Socket socket : mocked.constructed()) { + Mockito.verify(socket).connect(Mockito.any(InetSocketAddress.class), Mockito.eq(250)); + } + } finally { + timedRegistry.destroy(); + } + } - assertTrue(expired, "an unreachable provider must be treated as expired"); - assertTrue(elapsedMillis < 20_000, "isExpired must honour the connect timeout, took " + elapsedMillis + "ms"); + @Test + void testLivenessProbeUsesTheDefaultTimeoutWhenNotConfigured() throws Exception { + try (MockedConstruction mocked = Mockito.mockConstruction(Socket.class)) { + registry.isExpired(URL.valueOf("dubbo://127.0.0.1:20880/" + service)); + + assertFalse(mocked.constructed().isEmpty(), "the probe must attempt a connect"); + for (Socket socket : mocked.constructed()) { + Mockito.verify(socket) + .connect( + Mockito.any(InetSocketAddress.class), + Mockito.eq(Constants.DEFAULT_CLEAN_CHECK_TIMEOUT)); + } + } } @Test @@ -310,18 +340,24 @@ void testIsExpiredSkipsNonDynamicAndSpecialProtocols() { * than swallowing the InterruptedException. */ @Test - void testIsExpiredRestoresInterruptDuringRetrySleep() { - URL unreachable = URL.valueOf("dubbo://192.0.2.1:20880/" + service + "?timeout=200"); - Thread.currentThread().interrupt(); - try { - boolean expired = registry.isExpired(unreachable); - assertTrue(expired, "an unreachable provider must be treated as expired"); - assertTrue( - Thread.currentThread().isInterrupted(), - "the interrupt flag must be restored after the retry sleep"); - } finally { - // clear the flag so it cannot leak into other tests - Thread.interrupted(); + void testIsExpiredRestoresInterruptDuringRetrySleep() throws Exception { + URL unreachable = URL.valueOf("dubbo://127.0.0.1:20880/" + service); + // Force both probe attempts to fail without depending on the network, so the only reason + // the probe reaches the retry sleep is the connect failure itself. + try (MockedConstruction mocked = Mockito.mockConstruction( + Socket.class, (mock, context) -> Mockito.doThrow(new ConnectException("refused")) + .when(mock) + .connect(Mockito.any(InetSocketAddress.class), Mockito.anyInt()))) { + Thread.currentThread().interrupt(); + try { + assertTrue(registry.isExpired(unreachable), "a provider that cannot be reached is expired"); + assertTrue( + Thread.currentThread().isInterrupted(), + "the interrupt flag must be restored after the retry sleep"); + } finally { + // clear the flag so it cannot leak into other tests + Thread.interrupted(); + } } } }