diff --git a/core/src/main/java/tech/ydb/core/impl/pool/PriorityPicker.java b/core/src/main/java/tech/ydb/core/impl/pool/PriorityPicker.java index 185557261..b2d576f50 100644 --- a/core/src/main/java/tech/ydb/core/impl/pool/PriorityPicker.java +++ b/core/src/main/java/tech/ydb/core/impl/pool/PriorityPicker.java @@ -90,14 +90,25 @@ static String detectLocalDC(List endpoints, Ticker ticker) { int nodeSize = Math.min(nodes.size(), DETECT_DC_NODE_SIZE); long tcpPing = 0; + int reachableCount = 0; for (EndpointRecord node : nodes.subList(0, nodeSize)) { - long currentPing = tcpPing(new InetSocketAddress(node.getHost(), node.getPort()), ticker); - logger.debug("Address: {}, port: {}, nanos ping: {}", node.getHost(), node.getPort(), currentPing); - tcpPing += currentPing; + try { + long currentPing = tcpPing(new InetSocketAddress(node.getHost(), node.getPort()), ticker); + logger.debug("{}, nanos ping: {}", node, currentPing); + tcpPing += currentPing; + reachableCount++; + } catch (IOException ex) { + logger.warn("{} is not reachable with error {}", node, ex.getMessage()); + } } - tcpPing /= nodeSize; + if (reachableCount == 0) { + logger.debug("No reachable nodes in {}, skip it", dc); + continue; + } + + tcpPing /= reachableCount; if (minPing > tcpPing) { minPing = tcpPing; @@ -108,14 +119,11 @@ static String detectLocalDC(List endpoints, Ticker ticker) { return localDC; } - private static long tcpPing(InetSocketAddress socketAddress, Ticker ticker) { + private static long tcpPing(InetSocketAddress socketAddress, Ticker ticker) throws IOException { try (Socket socket = SocketFactory.getDefault().createSocket()) { - final long startConnection = ticker.read(); + long started = ticker.read(); socket.connect(socketAddress, DETECT_DC_TCP_PING_TIMEOUT_MS); - final long stopConnection = ticker.read(); - return stopConnection - startConnection; - } catch (IOException e) { - return Long.MAX_VALUE; + return ticker.read() - started; } } } diff --git a/core/src/test/java/tech/ydb/core/impl/pool/PriorityPickerTest.java b/core/src/test/java/tech/ydb/core/impl/pool/PriorityPickerTest.java index b2abd3cbf..c11cfdb8c 100644 --- a/core/src/test/java/tech/ydb/core/impl/pool/PriorityPickerTest.java +++ b/core/src/test/java/tech/ydb/core/impl/pool/PriorityPickerTest.java @@ -2,6 +2,7 @@ import java.io.IOException; import java.net.ServerSocket; +import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.stream.Collectors; @@ -108,4 +109,37 @@ public void detectLocalDCTest() { throw new RuntimeException(e); } } + + @Test + public void detectLocalDCWithUnreachableNodesTest() throws IOException { + int closedPort; + + try (ServerSocket socket = ServerSocketFactory.getDefault().createServerSocket(0)) { + closedPort = socket.getLocalPort(); + } + + // every successful probe takes 10 nanos + Ticker ticker = new Ticker() { + private long now = 0; + + @Override + public long read() { + now += 10; + return now; + } + }; + + try (ServerSocket serverSocket = ServerSocketFactory.getDefault().createServerSocket(0)) { + int openPort = serverSocket.getLocalPort(); + + List records = Arrays.asList( + new EndpointRecord("localhost", closedPort, 1, "DOWN", null), + new EndpointRecord("localhost", closedPort, 2, "DOWN", null), + new EndpointRecord("localhost", openPort, 3, "UP", null) + ); + + // failed probes of DOWN must not overflow into a negative, i.e. the best, ping + Assert.assertEquals("UP", PriorityPicker.detectLocalDC(records, ticker)); + } + } }