From 9df806068bd05ea6acddfa8a4f5148a4b2453ecc Mon Sep 17 00:00:00 2001 From: Aleksandr Iuskasov Date: Wed, 21 Oct 2026 21:38:14 +0500 Subject: [PATCH 1/8] fix(core): report greeting timeout as a timeout When a server accepts the TCP connection but does not send the greeting, the greeting timer closed the channel without setting a close reason. The close handler then treated it as a close by the server: it fired CLOSE_BY_REMOTE with "Connection closed by server" next to the TimeoutException of the connect future. The pool listens to both, so every failed attempt was handled twice and the reconnect chains multiplied: in one run of the new pool test, 12 attempts to a silent server scheduled 66 reconnects instead of one per attempt. Fail the connect promise with TimeoutException first and close the channel only if that succeeded, so a connection whose greeting has just arrived is never closed by the timer. Close it with a dedicated reason and do not fire a close event for it. Name the address and the timeout in the message. The timer also stayed armed when the server closed the connection before the greeting, and then fired during the next attempt and closed its channel: in two runs of the new test the second attempt was closed 220 and 326 ms after it started, with a 600 ms timeout. Let the timer close the channel only of the attempt it was scheduled for. --- CHANGELOG.md | 6 + .../core/connection/ConnectionImpl.java | 56 ++++++- .../core/connection/GreetingTimeoutTest.java | 158 ++++++++++++++++++ .../ConnectionToTarantoolTest.java | 6 +- .../pool/unit/ReconnectLoopTest.java | 147 ++++++++++++++++ 5 files changed, 361 insertions(+), 12 deletions(-) create mode 100644 tarantool-core/src/test/java/io/tarantool/core/connection/GreetingTimeoutTest.java create mode 100644 tarantool-pooling/src/test/java/io/tarantool/pool/unit/ReconnectLoopTest.java diff --git a/CHANGELOG.md b/CHANGELOG.md index de4ff77c..052c11e1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,12 @@ terminates with an error. Providing not positive timeout value throws an exception. - Fixed connection close callbacks executing before the connection state becomes closed. - Fixed greeting object remaining non-null after connection closure. +- Fixed a bug when a server that accepts the TCP connection but does not send the greeting was + reported as "Connection closed by server". The connect future is now failed only with + `TimeoutException`, whose message names the address and the timeout, and no + `CLOSE_BY_REMOTE` event is fired. Previously the connection pool handled every such attempt + twice, so the reconnect chains multiplied while the server stayed silent. The greeting timer of + an attempt that the server closed no longer closes the channel of the next attempt. ### Documentation diff --git a/tarantool-core/src/main/java/io/tarantool/core/connection/ConnectionImpl.java b/tarantool-core/src/main/java/io/tarantool/core/connection/ConnectionImpl.java index b3d31e40..2bac8c97 100644 --- a/tarantool-core/src/main/java/io/tarantool/core/connection/ConnectionImpl.java +++ b/tarantool-core/src/main/java/io/tarantool/core/connection/ConnectionImpl.java @@ -44,7 +44,9 @@ private enum CloseReason { /** Closed by client close reason. */ CLOSED_BY_CLIENT(ERR_CLOSED_BY_CLIENT), /** Closed by shutdown close reason. */ - CLOSED_BY_SHUTDOWN(ERR_CLOSED_BY_SHUTDOWN); + CLOSED_BY_SHUTDOWN(ERR_CLOSED_BY_SHUTDOWN), + /** Closed by the client because the greeting did not arrive in time. */ + CLOSED_BY_TIMEOUT(ERR_CLOSED_BY_TIMEOUT); private final String message; @@ -98,6 +100,9 @@ public String getMessage() { private static final String ERR_CLOSED_BY_CLIENT = "Connection closed by client"; private static final String ERR_CLOSED_BY_SERVER = "Connection closed by server"; private static final String ERR_CLOSED_BY_SHUTDOWN = "Connection closed by shutdown"; + private static final String ERR_CLOSED_BY_TIMEOUT = "Connection closed by connect timeout"; + private static final String ERR_GREETING_TIMEOUT = + "Connection timeout: no greeting from %s within %d ms"; private static final String ERR_NOT_CONNECTED = "Connection is not established"; private static final String ERR_SEND_FAILURE = "Failed to send IProto message: %s"; @@ -118,6 +123,10 @@ public String getMessage() { private Consumer consumer; private CompletableFuture connectPromise; private Timeout timeoutHandler; + + /** Greeting promise of the current attempt; a timer of an earlier attempt must not touch it. */ + private CompletableFuture greetingPromise; + private int idleTimeout; /** @@ -190,7 +199,8 @@ public synchronized CompletableFuture connect(InetSocketAddress addres log.debug("{} is connecting to {}", this, address); CompletableFuture promise = new CompletableFuture<>(); - scheduleGreetingTimeout(promise, timeoutMs); + greetingPromise = promise; + scheduleGreetingTimeout(promise, address, timeoutMs); connectPromise = promise.thenApply( greeting -> { @@ -358,6 +368,15 @@ private synchronized void onChannelClose(ChannelFuture f) { String message; ConnectionCloseEvent event; CloseReason lastCloseReason = closedBy.get(); + if (lastCloseReason == CloseReason.CLOSED_BY_TIMEOUT) { + // The greeting timer has already failed the connect promise with TimeoutException, so the + // greeting can no longer complete it: the connection has never been established and there + // is no close event to fire. + state.set(State.CLOSED); + channel = null; + connectPromise = null; + return; + } switch (lastCloseReason) { case CLOSED_BY_SHUTDOWN: message = lastCloseReason.getMessage(); @@ -416,23 +435,42 @@ private void fire(ConnectionCloseEvent event, Throwable ex) { } } - private void scheduleGreetingTimeout(CompletableFuture promise, long timeoutMs) { + private void scheduleGreetingTimeout( + CompletableFuture promise, InetSocketAddress address, long timeoutMs) { log.debug("Schedule timeout for connect in milliseconds: {}", timeoutMs); timeoutHandler = this.timerService.newTimeout( timeoutHandler -> { - if (promise.isDone()) { - return; - } - if (state.get() == State.CONNECTING || state.get() == State.WAITING) { - channel.pipeline().close(); - promise.completeExceptionally(new TimeoutException("Connection timeout")); + // Fail the promise before closing the channel: if the greeting has already + // completed it, the connection is established and must not be closed here. + if (promise.completeExceptionally( + new TimeoutException(String.format(ERR_GREETING_TIMEOUT, address, timeoutMs)))) { + closeOnGreetingTimeout(promise); } }, timeoutMs, TimeUnit.MILLISECONDS); } + /** + * Closes the channel of a connection whose greeting promise has been failed by the timeout. + * + *

The close reason is set before the channel is closed, so the close is not reported as a + * close by server: the connection has never been established. A timer of an earlier attempt does + * nothing: its channel is already closed, and the current one belongs to another attempt. + * + * @param promise greeting promise of the attempt the timer was scheduled for + */ + private synchronized void closeOnGreetingTimeout(CompletableFuture promise) { + if (promise != greetingPromise) { + return; + } + if (state.get() != State.CONNECTING && state.get() != State.WAITING) { + return; + } + closeChannel(CloseReason.CLOSED_BY_TIMEOUT); + } + private void cancelGreetingTimeout() { if (timeoutHandler == null) { return; diff --git a/tarantool-core/src/test/java/io/tarantool/core/connection/GreetingTimeoutTest.java b/tarantool-core/src/test/java/io/tarantool/core/connection/GreetingTimeoutTest.java new file mode 100644 index 00000000..5ea3e64f --- /dev/null +++ b/tarantool-core/src/test/java/io/tarantool/core/connection/GreetingTimeoutTest.java @@ -0,0 +1,158 @@ +/* + * Copyright (c) 2026 VK DIGITAL TECHNOLOGIES LIMITED LIABILITY COMPANY + * All Rights Reserved. + */ + +package io.tarantool.core.connection; + +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicLong; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import io.netty.bootstrap.Bootstrap; +import io.netty.channel.ChannelOption; +import io.netty.channel.EventLoopGroup; +import io.netty.channel.MultiThreadIoEventLoopGroup; +import io.netty.channel.nio.NioIoHandler; +import io.netty.channel.socket.nio.NioSocketChannel; +import io.netty.util.HashedWheelTimer; +import io.netty.util.Timer; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +/** A server accepts the TCP connection but never sends the greeting. */ +public class GreetingTimeoutTest { + + private static final long CONNECT_TIMEOUT_MS = 300; + private static final long EVENT_WAIT_MS = 1000; + + private EventLoopGroup group; + private Timer timer; + private ServerSocket silentServer; + private ConnectionImpl connection; + + @BeforeEach + public void setUp() throws Exception { + group = new MultiThreadIoEventLoopGroup(1, NioIoHandler.newFactory()); + timer = new HashedWheelTimer(); + // The kernel completes the TCP handshake from the backlog; nobody reads or writes. + silentServer = new ServerSocket(0, 50, InetAddress.getLoopbackAddress()); + Bootstrap bootstrap = + new Bootstrap() + .group(group) + .channel(NioSocketChannel.class) + .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000); + connection = new ConnectionImpl(bootstrap, timer); + } + + @AfterEach + public void tearDown() throws Exception { + connection.close(); + silentServer.close(); + timer.stop(); + group.shutdownGracefully().get(); + } + + @Test + public void test_greetingTimeout_shouldNotBeReportedAsCloseByServer() throws Exception { + List closedByRemote = new CopyOnWriteArrayList<>(); + connection.onClose(ConnectionCloseEvent.CLOSE_BY_REMOTE, (c, ex) -> closedByRemote.add(ex)); + + InetSocketAddress address = + new InetSocketAddress(silentServer.getInetAddress(), silentServer.getLocalPort()); + CompletableFuture connect = connection.connect(address, CONNECT_TIMEOUT_MS); + + ExecutionException ex = + assertThrows(ExecutionException.class, () -> connect.get(5, TimeUnit.SECONDS)); + assertInstanceOf(TimeoutException.class, ex.getCause()); + assertEquals( + String.format( + "Connection timeout: no greeting from %s within %d ms", address, CONNECT_TIMEOUT_MS), + ex.getCause().getMessage()); + + // Without the fix the close event follows the timeout within milliseconds. + long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(EVENT_WAIT_MS); + while (System.nanoTime() < deadline && closedByRemote.isEmpty()) { + Thread.sleep(10); + } + assertEquals( + 0, + closedByRemote.size(), + "greeting timeout is reported as closed by server: " + closedByRemote); + assertFalse(connection.isConnected()); + } + + @Test + public void test_greetingTimerOfClosedAttempt_shouldNotCloseNextAttempt() throws Exception { + long timeoutMs = 600; + long pauseMs = 300; + try (ServerSocket closingServer = new ServerSocket(0, 50, InetAddress.getLoopbackAddress())) { + // Accepts and closes at once, before the greeting. + Thread closer = + new Thread( + () -> { + try (Socket socket = closingServer.accept()) { + // closed by try-with-resources + } catch (Exception ignored) { + // the test fails on its own assertions + } + }); + closer.setDaemon(true); + closer.start(); + + CompletableFuture first = + connection.connect( + new InetSocketAddress(closingServer.getInetAddress(), closingServer.getLocalPort()), + timeoutMs); + assertThrows(ExecutionException.class, () -> first.get(5, TimeUnit.SECONDS)); + } + Thread.sleep(pauseMs); + + // The silent server records when the client closed its side of the second attempt. + AtomicLong closedAfterMs = new AtomicLong(-1); + long start = System.nanoTime(); + Thread reader = + new Thread( + () -> { + try (Socket socket = silentServer.accept()) { + while (socket.getInputStream().read() != -1) { + // the client never writes + } + } catch (Exception ignored) { + // a reset also ends the connection + } + closedAfterMs.set(TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start)); + }); + reader.setDaemon(true); + reader.start(); + + CompletableFuture second = + connection.connect( + new InetSocketAddress(silentServer.getInetAddress(), silentServer.getLocalPort()), + timeoutMs); + ExecutionException ex = + assertThrows(ExecutionException.class, () -> second.get(5, TimeUnit.SECONDS)); + assertInstanceOf(TimeoutException.class, ex.getCause()); + reader.join(5000); + + // Without the fix the timer of the first attempt closes the second one about timeoutMs - + // pauseMs after it starts, without any error for it. + assertTrue( + closedAfterMs.get() >= timeoutMs - 100, + "second attempt was closed after " + closedAfterMs.get() + " ms"); + } +} diff --git a/tarantool-core/src/test/java/io/tarantool/core/integration/ConnectionToTarantoolTest.java b/tarantool-core/src/test/java/io/tarantool/core/integration/ConnectionToTarantoolTest.java index 77ea1ee4..187d3ea7 100644 --- a/tarantool-core/src/test/java/io/tarantool/core/integration/ConnectionToTarantoolTest.java +++ b/tarantool-core/src/test/java/io/tarantool/core/integration/ConnectionToTarantoolTest.java @@ -249,7 +249,7 @@ public void testConnectToAddressWithBadHost() { cause.getMessage()); assertEquals(ConnectTimeoutException.class, findRootCause(ex).getClass()); } else if (cause.getClass() == TimeoutException.class) { - assertEquals("Connection timeout", cause.getMessage()); + assertTrue(cause.getMessage().startsWith("Connection timeout"), cause.getMessage()); log.info(findRootCause(ex).getClass().toString()); // Need for debug in the future } else { fail("An unknown exception was thrown during connection to the bad hostname!", cause); @@ -322,7 +322,7 @@ public void testConnectToSilentNode() { Exception ex = assertThrows(CompletionException.class, future::join); Throwable cause = findRootCause(ex); assertEquals(TimeoutException.class, cause.getClass()); - assertEquals("Connection timeout", cause.getMessage()); + assertTrue(cause.getMessage().startsWith("Connection timeout"), cause.getMessage()); } /* @@ -408,7 +408,7 @@ public void testConnectWithGreetingTimeout() throws InterruptedException { assertTrue( cause instanceof TimeoutException || cause instanceof ConnectionClosedException, "Exception should be either TimeoutException or ConnectionClosedException"); - assertEquals("Connection timeout", cause.getMessage()); + assertTrue(cause.getMessage().startsWith("Connection timeout"), cause.getMessage()); } @Test diff --git a/tarantool-pooling/src/test/java/io/tarantool/pool/unit/ReconnectLoopTest.java b/tarantool-pooling/src/test/java/io/tarantool/pool/unit/ReconnectLoopTest.java new file mode 100644 index 00000000..28e57050 --- /dev/null +++ b/tarantool-pooling/src/test/java/io/tarantool/pool/unit/ReconnectLoopTest.java @@ -0,0 +1,147 @@ +/* + * Copyright (c) 2026 VK DIGITAL TECHNOLOGIES LIMITED LIABILITY COMPANY + * All Rights Reserved. + */ + +package io.tarantool.pool.unit; + +import java.io.IOException; +import java.net.InetAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.assertTrue; +import io.micrometer.core.instrument.MeterRegistry; +import io.netty.bootstrap.Bootstrap; +import io.netty.channel.ChannelOption; +import io.netty.channel.EventLoopGroup; +import io.netty.channel.MultiThreadIoEventLoopGroup; +import io.netty.channel.nio.NioIoHandler; +import io.netty.channel.socket.nio.NioSocketChannel; +import io.netty.util.HashedWheelTimer; +import io.netty.util.Timer; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import io.tarantool.core.ManagedResource; +import io.tarantool.core.connection.ConnectionFactory; +import io.tarantool.pool.IProtoClientPoolImpl; +import io.tarantool.pool.InstanceConnectionGroup; +import io.tarantool.pool.PoolEventListener; + +/** A pool entry keeps reconnecting to a server that accepts connections and never answers. */ +public class ReconnectLoopTest { + + private static final String TAG = "silent"; + private static final long CONNECT_TIMEOUT_MS = 200; + private static final long RECONNECT_AFTER_MS = 50; + private static final long OBSERVE_MS = 3_000; + + private EventLoopGroup group; + private Timer timer; + private ServerSocket server; + private Thread acceptor; + private final List accepted = Collections.synchronizedList(new ArrayList<>()); + private IProtoClientPoolImpl pool; + + @BeforeEach + public void setUp() throws IOException { + group = new MultiThreadIoEventLoopGroup(1, NioIoHandler.newFactory()); + timer = new HashedWheelTimer(10, TimeUnit.MILLISECONDS); + server = new ServerSocket(0, 50, InetAddress.getLoopbackAddress()); + // Accepts connections and never writes the greeting; each accepted socket is one attempt. + acceptor = + new Thread( + () -> { + while (!server.isClosed()) { + try { + accepted.add(server.accept()); + } catch (IOException e) { + return; + } + } + }); + acceptor.setDaemon(true); + acceptor.start(); + } + + @AfterEach + public void tearDown() throws Exception { + if (pool != null) { + pool.close(); + } + server.close(); + synchronized (accepted) { + for (Socket socket : accepted) { + socket.close(); + } + } + timer.stop(); + group.shutdownGracefully().get(); + } + + @Test + public void test_failedAttempts_shouldNotMultiplyReconnectTasks() throws Exception { + AtomicInteger scheduled = new AtomicInteger(); + PoolEventListener listener = + new PoolEventListener() { + @Override + public void onReconnectScheduled(String tag, int index, long delayMs) { + scheduled.incrementAndGet(); + } + }; + pool = createPool(null, listener); + pool.get(TAG, 0); + + Thread.sleep(OBSERVE_MS); + + int attempts = accepted.size(); + assertTrue(attempts >= 3, "too few attempts to judge: " + attempts); + // Each failed attempt schedules one reconnect. When a greeting timeout was also reported as + // a close by server, every attempt was handled twice and the reconnect chains multiplied. + assertTrue( + scheduled.get() <= attempts + 1, + "reconnect tasks multiply: " + + scheduled.get() + + " scheduled for " + + attempts + + " attempts"); + } + + private IProtoClientPoolImpl createPool(MeterRegistry registry, PoolEventListener listener) { + Bootstrap bootstrap = + new Bootstrap() + .group(group) + .channel(NioSocketChannel.class) + .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000); + IProtoClientPoolImpl pool = + new IProtoClientPoolImpl( + new ConnectionFactory(bootstrap, timer), + ManagedResource.external(timer), + true, + null, + null, + registry, + null, + null, + false, + listener); + pool.setConnectTimeout(CONNECT_TIMEOUT_MS); + pool.setReconnectAfter(RECONNECT_AFTER_MS); + pool.setGroups( + Collections.singletonList( + InstanceConnectionGroup.builder() + .withHost(server.getInetAddress().getHostAddress()) + .withPort(server.getLocalPort()) + .withSize(1) + .withTag(TAG) + .build())); + return pool; + } +} From de1bd9d36215829c0fcf45f46e69a420169a91d5 Mon Sep 17 00:00:00 2001 From: Aleksandr Iuskasov Date: Thu, 22 Oct 2026 21:36:56 +0500 Subject: [PATCH 2/8] fix(core): skip close event before the greeting When a server, or a proxy in front of it, accepted the TCP connection and closed it before the greeting, the close was reported twice: the connect future failed with ConnectionClosedException, and the close handler also fired CLOSE_BY_REMOTE. The pool listens to both and scheduled a reconnect for each, so the reconnect chains multiplied: in one run of the new pool test, 50 attempts in 3 s scheduled 1325 reconnects. A connection that has not received the greeting has no requests in flight and no registered watchers, and its failure already reaches the caller through the connect future. Fire close events only for a connection that has received the greeting. --- CHANGELOG.md | 4 ++ .../core/connection/ConnectionImpl.java | 11 +++- .../pool/unit/ReconnectLoopTest.java | 50 ++++++++++++++++++- 3 files changed, 62 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 052c11e1..4bc47073 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,6 +21,10 @@ `CLOSE_BY_REMOTE` event is fired. Previously the connection pool handled every such attempt twice, so the reconnect chains multiplied while the server stayed silent. The greeting timer of an attempt that the server closed no longer closes the channel of the next attempt. +- Fixed reconnect chains multiplying when a server (or a proxy in front of it) accepts the TCP + connection and closes it before the greeting: the close was reported both by the failed connect + future and by a `CLOSE_BY_REMOTE` event, and the pool handled each report. Close events are now + fired only for a connection that has received the greeting. ### Documentation diff --git a/tarantool-core/src/main/java/io/tarantool/core/connection/ConnectionImpl.java b/tarantool-core/src/main/java/io/tarantool/core/connection/ConnectionImpl.java index 2bac8c97..15f31ba0 100644 --- a/tarantool-core/src/main/java/io/tarantool/core/connection/ConnectionImpl.java +++ b/tarantool-core/src/main/java/io/tarantool/core/connection/ConnectionImpl.java @@ -127,6 +127,12 @@ public String getMessage() { /** Greeting promise of the current attempt; a timer of an earlier attempt must not touch it. */ private CompletableFuture greetingPromise; + /** + * Whether the current attempt has received the greeting. A close of a connection that has never + * been established is reported only through the failed connect future, not by a close event. + */ + private volatile boolean established; + private int idleTimeout; /** @@ -200,6 +206,7 @@ public synchronized CompletableFuture connect(InetSocketAddress addres CompletableFuture promise = new CompletableFuture<>(); greetingPromise = promise; + established = false; scheduleGreetingTimeout(promise, address, timeoutMs); connectPromise = promise.thenApply( @@ -207,6 +214,7 @@ public synchronized CompletableFuture connect(InetSocketAddress addres cancelGreetingTimeout(); log.info("{} connected to {}", this, address); this.greeting = greeting; + established = true; state.set(State.READY); return greeting; }); @@ -396,13 +404,14 @@ private synchronized void onChannelClose(ChannelFuture f) { if (notify) { failConnectPromise(ex); } + final boolean fireEvent = notify && established; channel = null; connectPromise = null; greeting = null; state.set(State.CLOSED); - if (notify) { + if (fireEvent) { fire(event, ex); } } diff --git a/tarantool-pooling/src/test/java/io/tarantool/pool/unit/ReconnectLoopTest.java b/tarantool-pooling/src/test/java/io/tarantool/pool/unit/ReconnectLoopTest.java index 28e57050..8652852c 100644 --- a/tarantool-pooling/src/test/java/io/tarantool/pool/unit/ReconnectLoopTest.java +++ b/tarantool-pooling/src/test/java/io/tarantool/pool/unit/ReconnectLoopTest.java @@ -114,7 +114,53 @@ public void onReconnectScheduled(String tag, int index, long delayMs) { + " attempts"); } + @Test + public void test_closedBeforeGreeting_shouldNotMultiplyReconnectTasks() throws Exception { + AtomicInteger scheduled = new AtomicInteger(); + PoolEventListener listener = + new PoolEventListener() { + @Override + public void onReconnectScheduled(String tag, int index, long delayMs) { + scheduled.incrementAndGet(); + } + }; + AtomicInteger attempts = new AtomicInteger(); + try (ServerSocket closingServer = new ServerSocket(0, 50, InetAddress.getLoopbackAddress())) { + // Accepts and closes at once, before the greeting, like a proxy without a backend. + Thread closer = + new Thread( + () -> { + while (!closingServer.isClosed()) { + try (Socket socket = closingServer.accept()) { + attempts.incrementAndGet(); + } catch (IOException e) { + return; + } + } + }); + closer.setDaemon(true); + closer.start(); + pool = createPool(null, listener, closingServer); + pool.get(TAG, 0); + + Thread.sleep(OBSERVE_MS); + } + + int count = attempts.get(); + assertTrue(count >= 3, "too few attempts to judge: " + count); + // A connection closed before the greeting has never been established: the failed connect + // future is the only report of the attempt, so each attempt schedules one reconnect. + assertTrue( + scheduled.get() <= count + 1, + "reconnect tasks multiply: " + scheduled.get() + " scheduled for " + count + " attempts"); + } + private IProtoClientPoolImpl createPool(MeterRegistry registry, PoolEventListener listener) { + return createPool(registry, listener, server); + } + + private IProtoClientPoolImpl createPool( + MeterRegistry registry, PoolEventListener listener, ServerSocket target) { Bootstrap bootstrap = new Bootstrap() .group(group) @@ -137,8 +183,8 @@ private IProtoClientPoolImpl createPool(MeterRegistry registry, PoolEventListene pool.setGroups( Collections.singletonList( InstanceConnectionGroup.builder() - .withHost(server.getInetAddress().getHostAddress()) - .withPort(server.getLocalPort()) + .withHost(target.getInetAddress().getHostAddress()) + .withPort(target.getLocalPort()) .withSize(1) .withTag(TAG) .build())); From 76be67dda67ca56cbb6a0604ab7e8a7ed22df5fe Mon Sep 17 00:00:00 2001 From: Aleksandr Iuskasov Date: Wed, 21 Oct 2026 21:39:35 +0500 Subject: [PATCH 3/8] fix(core): register box.shutdown callback once With graceful shutdown enabled, IProtoClientImpl.connect() called watch("box.shutdown", this::shutdownEventCallback) on every attempt. watch() appends the callback to the existing watcher and nothing removes it, so a client that kept reconnecting to an unavailable server accumulated one callback per attempt, and the event after a successful connect ran the callback once per previous attempt. The watcher itself survives reconnects and is re-subscribed by updateWatchers(), so add the callback only on the first connect. --- CHANGELOG.md | 4 + .../io/tarantool/core/IProtoClientImpl.java | 7 +- .../core/ShutdownWatcherRegistrationTest.java | 90 +++++++++++++++++++ 3 files changed, 100 insertions(+), 1 deletion(-) create mode 100644 tarantool-core/src/test/java/io/tarantool/core/ShutdownWatcherRegistrationTest.java diff --git a/CHANGELOG.md b/CHANGELOG.md index 4bc47073..7b2f8e55 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,6 +25,10 @@ connection and closes it before the greeting: the close was reported both by the failed connect future and by a `CLOSE_BY_REMOTE` event, and the pool handled each report. Close events are now fired only for a connection that has received the greeting. +- Fixed a leak when graceful shutdown is enabled: every connect attempt added one more + `box.shutdown` callback to the same watcher, so a client that kept reconnecting accumulated + callbacks, and a `box.shutdown` event after a successful connect ran the callback once per + previous attempt. The callback is now added once per client. ### Documentation diff --git a/tarantool-core/src/main/java/io/tarantool/core/IProtoClientImpl.java b/tarantool-core/src/main/java/io/tarantool/core/IProtoClientImpl.java index b248c732..fa8c213c 100644 --- a/tarantool-core/src/main/java/io/tarantool/core/IProtoClientImpl.java +++ b/tarantool-core/src/main/java/io/tarantool/core/IProtoClientImpl.java @@ -16,6 +16,7 @@ import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; import java.util.function.BiConsumer; import java.util.function.Consumer; @@ -97,6 +98,10 @@ public class IProtoClientImpl implements IProtoClient { private final List toUnwatch; protected final Map fsmRegistry; private final Map watchers; + + /** Set once the {@code box.shutdown} callback is added; it survives reconnects. */ + private final AtomicBoolean shutdownWatcherRegistered = new AtomicBoolean(); + protected final Timer timerService; private final WatcherOptions watcherOpts; private CompletableFuture serverProtocolVersion; @@ -185,7 +190,7 @@ public CompletableFuture connect(InetSocketAddress address, long timeoutMs @Override public CompletableFuture connect( InetSocketAddress address, long timeoutMs, boolean gracefulShutdown) { - if (gracefulShutdown) { + if (gracefulShutdown && shutdownWatcherRegistered.compareAndSet(false, true)) { watch(SHUTDOWN_EVENT_KEY, this::shutdownEventCallback); } diff --git a/tarantool-core/src/test/java/io/tarantool/core/ShutdownWatcherRegistrationTest.java b/tarantool-core/src/test/java/io/tarantool/core/ShutdownWatcherRegistrationTest.java new file mode 100644 index 00000000..d4d68d7f --- /dev/null +++ b/tarantool-core/src/test/java/io/tarantool/core/ShutdownWatcherRegistrationTest.java @@ -0,0 +1,90 @@ +/* + * Copyright (c) 2026 VK DIGITAL TECHNOLOGIES LIMITED LIABILITY COMPANY + * All Rights Reserved. + */ + +package io.tarantool.core; + +import java.lang.reflect.Field; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.ServerSocket; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import io.netty.bootstrap.Bootstrap; +import io.netty.channel.ChannelOption; +import io.netty.channel.EventLoopGroup; +import io.netty.channel.MultiThreadIoEventLoopGroup; +import io.netty.channel.nio.NioIoHandler; +import io.netty.channel.socket.nio.NioSocketChannel; +import io.netty.util.HashedWheelTimer; +import io.netty.util.Timer; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import io.tarantool.core.connection.ConnectionFactory; + +/** Reconnect attempts must not add a new {@code box.shutdown} callback each time. */ +public class ShutdownWatcherRegistrationTest { + + private static final int ATTEMPTS = 5; + + private EventLoopGroup group; + private Timer timer; + private IProtoClientImpl client; + + @BeforeEach + public void setUp() { + group = new MultiThreadIoEventLoopGroup(1, NioIoHandler.newFactory()); + timer = new HashedWheelTimer(); + Bootstrap bootstrap = + new Bootstrap() + .group(group) + .channel(NioSocketChannel.class) + .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000); + client = new IProtoClientImpl(new ConnectionFactory(bootstrap, timer), timer); + } + + @AfterEach + public void tearDown() throws Exception { + client.close(); + timer.stop(); + group.shutdownGracefully().get(); + } + + @Test + public void test_failedConnects_shouldRegisterShutdownCallbackOnce() throws Exception { + InetSocketAddress refused = closedPortAddress(); + + for (int i = 0; i < ATTEMPTS; i++) { + CompletableFuture attempt = client.connect(refused, 1000, true); + assertThrows(ExecutionException.class, () -> attempt.get(5, TimeUnit.SECONDS)); + } + + assertEquals(1, shutdownCallbacksCount()); + } + + private static InetSocketAddress closedPortAddress() throws Exception { + try (ServerSocket socket = new ServerSocket(0, 1, InetAddress.getLoopbackAddress())) { + return new InetSocketAddress(socket.getInetAddress(), socket.getLocalPort()); + } + } + + // The callbacks are not exposed by the API, so the test reads them directly. + @SuppressWarnings("unchecked") + private int shutdownCallbacksCount() throws Exception { + Field watchersField = IProtoClientImpl.class.getDeclaredField("watchers"); + watchersField.setAccessible(true); + Map watchers = (Map) watchersField.get(client); + Field callbacksField = Watcher.class.getDeclaredField("callbacks"); + callbacksField.setAccessible(true); + return ((List) callbacksField.get(watchers.get("box.shutdown"))).size(); + } +} From 4d7fe01bba5d374a0e1a128368dc9f293c1a6e2c Mon Sep 17 00:00:00 2001 From: Aleksandr Iuskasov Date: Thu, 22 Oct 2026 21:17:03 +0500 Subject: [PATCH 4/8] fix(pool): stop connect timer on failed attempt The pool.connect.time sample was stopped only when the connect future succeeded, so every failed attempt left an active sample behind: in the new test, 12 attempts to a silent server left 12 active samples. Stop the sample when the connect future completes either way. --- CHANGELOG.md | 2 ++ .../java/io/tarantool/pool/PoolEntry.java | 3 +- .../pool/unit/ReconnectLoopTest.java | 34 +++++++++++++++++++ 3 files changed, 38 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 7b2f8e55..c64a35ee 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -29,6 +29,8 @@ `box.shutdown` callback to the same watcher, so a client that kept reconnecting accumulated callbacks, and a `box.shutdown` event after a successful connect ran the callback once per previous attempt. The callback is now added once per client. +- Fixed the `pool.connect.time` timer of the connection pool: it was stopped only when an + attempt succeeded, so every failed attempt left an active sample behind. ### Documentation diff --git a/tarantool-pooling/src/main/java/io/tarantool/pool/PoolEntry.java b/tarantool-pooling/src/main/java/io/tarantool/pool/PoolEntry.java index ef8b34ad..61ab1076 100644 --- a/tarantool-pooling/src/main/java/io/tarantool/pool/PoolEntry.java +++ b/tarantool-pooling/src/main/java/io/tarantool/pool/PoolEntry.java @@ -363,12 +363,13 @@ private CompletableFuture internalConnect() { this.metrics != null ? this.metrics.poolConnectTime.start() : null; CompletableFuture future = client.connect(group.getAddress(), connectTimeout, gracefulShutdown); + // Stop the timer on failure too, otherwise every failed attempt leaves an active sample. + future.whenComplete((r, exc) -> PoolEntryMetrics.stopTimer(timer)); String user = group.getUser(); connectFuture = future .thenCompose( greeting -> { - PoolEntryMetrics.stopTimer(timer); if (user != null) { return client.authorize(user, group.getPassword(), group.getAuthType()); } diff --git a/tarantool-pooling/src/test/java/io/tarantool/pool/unit/ReconnectLoopTest.java b/tarantool-pooling/src/test/java/io/tarantool/pool/unit/ReconnectLoopTest.java index 8652852c..c4779cae 100644 --- a/tarantool-pooling/src/test/java/io/tarantool/pool/unit/ReconnectLoopTest.java +++ b/tarantool-pooling/src/test/java/io/tarantool/pool/unit/ReconnectLoopTest.java @@ -16,7 +16,10 @@ import java.util.concurrent.atomic.AtomicInteger; import static org.junit.jupiter.api.Assertions.assertTrue; +import io.micrometer.core.instrument.Counter; +import io.micrometer.core.instrument.LongTaskTimer; import io.micrometer.core.instrument.MeterRegistry; +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; import io.netty.bootstrap.Bootstrap; import io.netty.channel.ChannelOption; import io.netty.channel.EventLoopGroup; @@ -155,6 +158,27 @@ public void onReconnectScheduled(String tag, int index, long delayMs) { "reconnect tasks multiply: " + scheduled.get() + " scheduled for " + count + " attempts"); } + @Test + public void test_failedAttempts_shouldStopConnectTimer() throws Exception { + MeterRegistry registry = createMetricsRegistry(); + pool = createPool(registry, null); + pool.get(TAG, 0); + + Thread.sleep(OBSERVE_MS); + + int attempts = accepted.size(); + assertTrue(attempts >= 3, "too few attempts to judge: " + attempts); + LongTaskTimer connectTime = registry.find("pool.connect.time").longTaskTimer(); + // At most one attempt of the single connection can be in flight. + assertTrue( + connectTime.activeTasks() <= 1, + "connect timer is not stopped on failure: " + + connectTime.activeTasks() + + " active for " + + attempts + + " attempts"); + } + private IProtoClientPoolImpl createPool(MeterRegistry registry, PoolEventListener listener) { return createPool(registry, listener, server); } @@ -190,4 +214,14 @@ private IProtoClientPoolImpl createPool( .build())); return pool; } + + private static MeterRegistry createMetricsRegistry() { + MeterRegistry registry = new SimpleMeterRegistry(); + LongTaskTimer.builder("request.timer").register(registry); + Counter.builder("request.counter").register(registry); + Counter.builder("response.success").register(registry); + Counter.builder("response.errors").register(registry); + Counter.builder("response.ignored").register(registry); + return registry; + } } From b55ac4c59a95588bf757b02e50db3a909b4e7acc Mon Sep 17 00:00:00 2001 From: Aleksandr Iuskasov Date: Thu, 22 Oct 2026 21:17:33 +0500 Subject: [PATCH 5/8] fix(pool): pass reconnectAfter to pool entries TarantoolClientImpl calls pool.setGroups() before pool.setReconnectAfter(), so the connections were created with the pool default of 1000 ms. setReconnectAfter() changed only the pool field, and PoolEntry kept its final copy, so withReconnectAfter() and the reconnect-after property of Spring Data had no effect on these connections. Pass the new value to existing connections, as setConnectTimeout() already does. --- CHANGELOG.md | 4 + .../tarantool/pool/IProtoClientPoolImpl.java | 2 + .../java/io/tarantool/pool/PoolEntry.java | 11 +- .../pool/unit/ReconnectDelayTest.java | 116 ++++++++++++++++++ 4 files changed, 132 insertions(+), 1 deletion(-) create mode 100644 tarantool-pooling/src/test/java/io/tarantool/pool/unit/ReconnectDelayTest.java diff --git a/CHANGELOG.md b/CHANGELOG.md index c64a35ee..cd1091f6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -31,6 +31,10 @@ previous attempt. The callback is now added once per client. - Fixed the `pool.connect.time` timer of the connection pool: it was stopped only when an attempt succeeded, so every failed attempt left an active sample behind. +- Fixed `reconnectAfter` being ignored for the connections created with the client: the client + builder set it on the pool after the connections had been created with the default of 1000 ms, + and the pool did not pass the new value to existing connections. `withReconnectAfter` and the + `reconnect-after` property of Spring Data now take effect. ### Documentation diff --git a/tarantool-pooling/src/main/java/io/tarantool/pool/IProtoClientPoolImpl.java b/tarantool-pooling/src/main/java/io/tarantool/pool/IProtoClientPoolImpl.java index a4f85cc5..f60880fe 100644 --- a/tarantool-pooling/src/main/java/io/tarantool/pool/IProtoClientPoolImpl.java +++ b/tarantool-pooling/src/main/java/io/tarantool/pool/IProtoClientPoolImpl.java @@ -501,6 +501,8 @@ public void setReconnectAfter(long reconnectAfter) throws IllegalArgumentExcepti if (reconnectAfter <= 0) { throw new IllegalArgumentException("reconnect period should be positive number"); } + entries.forEach( + (tag, entryGroup) -> entryGroup.forEach(pe -> pe.setReconnectAfter(reconnectAfter))); this.reconnectAfter = reconnectAfter; } diff --git a/tarantool-pooling/src/main/java/io/tarantool/pool/PoolEntry.java b/tarantool-pooling/src/main/java/io/tarantool/pool/PoolEntry.java index 61ab1076..26fc5062 100644 --- a/tarantool-pooling/src/main/java/io/tarantool/pool/PoolEntry.java +++ b/tarantool-pooling/src/main/java/io/tarantool/pool/PoolEntry.java @@ -97,7 +97,7 @@ final class PoolEntry { private final int index; /** Time in milliseconds after that reconnect task will be executed. */ - private final long reconnectAfter; + private volatile long reconnectAfter; /** Parameter that will say tarantool not to use Tuple Extension. */ private final boolean useTupleExtension; @@ -318,6 +318,15 @@ public void setConnectTimeout(long timeout) { this.connectTimeout = timeout; } + /** + * Setter for {@link #reconnectAfter}. + * + * @param reconnectAfter reconnect time in milliseconds + */ + public void setReconnectAfter(long reconnectAfter) { + this.reconnectAfter = reconnectAfter; + } + /** Heartbeat starter. */ public void startHeartbeat() { if (isHeartbeatStarted || heartbeatOpts == null) { diff --git a/tarantool-pooling/src/test/java/io/tarantool/pool/unit/ReconnectDelayTest.java b/tarantool-pooling/src/test/java/io/tarantool/pool/unit/ReconnectDelayTest.java new file mode 100644 index 00000000..84c86c31 --- /dev/null +++ b/tarantool-pooling/src/test/java/io/tarantool/pool/unit/ReconnectDelayTest.java @@ -0,0 +1,116 @@ +/* + * Copyright (c) 2026 VK DIGITAL TECHNOLOGIES LIMITED LIABILITY COMPANY + * All Rights Reserved. + */ + +package io.tarantool.pool.unit; + +import java.net.InetAddress; +import java.net.ServerSocket; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.TimeUnit; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import io.netty.bootstrap.Bootstrap; +import io.netty.channel.ChannelOption; +import io.netty.channel.EventLoopGroup; +import io.netty.channel.MultiThreadIoEventLoopGroup; +import io.netty.channel.nio.NioIoHandler; +import io.netty.channel.socket.nio.NioSocketChannel; +import io.netty.util.HashedWheelTimer; +import io.netty.util.Timer; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import io.tarantool.core.ManagedResource; +import io.tarantool.core.connection.ConnectionFactory; +import io.tarantool.pool.IProtoClientPoolImpl; +import io.tarantool.pool.InstanceConnectionGroup; +import io.tarantool.pool.PoolEventListener; + +/** Reconnect delays of a pool entry whose server refuses connections. */ +public class ReconnectDelayTest { + + private static final String TAG = "refused"; + + private EventLoopGroup group; + private Timer timer; + private int refusedPort; + private final List delays = Collections.synchronizedList(new ArrayList<>()); + private IProtoClientPoolImpl pool; + + @BeforeEach + public void setUp() throws Exception { + group = new MultiThreadIoEventLoopGroup(1, NioIoHandler.newFactory()); + timer = new HashedWheelTimer(10, TimeUnit.MILLISECONDS); + try (ServerSocket socket = new ServerSocket(0, 1, InetAddress.getLoopbackAddress())) { + refusedPort = socket.getLocalPort(); + } + Bootstrap bootstrap = + new Bootstrap() + .group(group) + .channel(NioSocketChannel.class) + .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000); + PoolEventListener listener = + new PoolEventListener() { + @Override + public void onReconnectScheduled(String tag, int index, long delayMs) { + delays.add(delayMs); + } + }; + pool = + new IProtoClientPoolImpl( + new ConnectionFactory(bootstrap, timer), + ManagedResource.external(timer), + true, + null, + null, + null, + null, + null, + false, + listener); + } + + @AfterEach + public void tearDown() throws Exception { + pool.close(); + timer.stop(); + group.shutdownGracefully().get(); + } + + @Test + public void test_reconnectAfter_shouldApplyToExistingConnections() throws Exception { + // TarantoolClientImpl sets the groups first and the reconnect delay after that. + setGroups(); + pool.setReconnectAfter(120); + pool.get(TAG, 0); + + awaitDelays(1); + + assertEquals(120L, delays.get(0)); + } + + private void setGroups() { + pool.setGroups( + Collections.singletonList( + InstanceConnectionGroup.builder() + .withHost(InetAddress.getLoopbackAddress().getHostAddress()) + .withPort(refusedPort) + .withSize(1) + .withTag(TAG) + .build())); + } + + private void awaitDelays(int count) throws InterruptedException { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10); + while (delays.size() < count && System.nanoTime() < deadline) { + Thread.sleep(10); + } + assertTrue(delays.size() >= count, "reconnects scheduled: " + delays); + } +} From 14ee525b2b5c3db6b75b076ff25fdf25f8c5a043 Mon Sep 17 00:00:00 2001 From: Aleksandr Iuskasov Date: Thu, 22 Oct 2026 21:20:46 +0500 Subject: [PATCH 6/8] fix(pool): make PoolEntry lock and unlock atomic lock() and unlock() checked a plain boolean and then changed it and the unavailable counter. They are called from the connect, close and heartbeat handlers, which run on different threads, so two calls could both pass the check and change the counter twice. In the new test, 8 threads locking and unlocking one entry left the counter at 4, -8 and -2 in three runs instead of 0. A counter below the real value makes hasAvailableClients() report a free connection while all entries are locked, so the balancer loops in getNext(); one above it hides free connections, and above the pool size the pool fails with "Invalid pool state". Keep the flag in an AtomicBoolean and change the counter only in the call that won the compare-and-set. --- CHANGELOG.md | 5 + .../java/io/tarantool/pool/PoolEntry.java | 16 +-- .../io/tarantool/pool/PoolEntryLockTest.java | 113 ++++++++++++++++++ 3 files changed, 126 insertions(+), 8 deletions(-) create mode 100644 tarantool-pooling/src/test/java/io/tarantool/pool/PoolEntryLockTest.java diff --git a/CHANGELOG.md b/CHANGELOG.md index cd1091f6..e73d6ec0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -31,6 +31,11 @@ previous attempt. The callback is now added once per client. - Fixed the `pool.connect.time` timer of the connection pool: it was stopped only when an attempt succeeded, so every failed attempt left an active sample behind. +- Fixed the count of unavailable connections in the pool drifting when a connection was locked + or unlocked from several threads at once (connect, close and heartbeat handlers): both calls + could pass the check and change the count twice. A count below the real value made the + balancer loop while all connections were locked; a count above it hid free connections or made + the pool fail with "Invalid pool state". - Fixed `reconnectAfter` being ignored for the connections created with the client: the client builder set it on the pool after the connections had been created with the default of 1000 ms, and the pool did not pass the new value to existing connections. `withReconnectAfter` and the diff --git a/tarantool-pooling/src/main/java/io/tarantool/pool/PoolEntry.java b/tarantool-pooling/src/main/java/io/tarantool/pool/PoolEntry.java index 26fc5062..a70e31e5 100644 --- a/tarantool-pooling/src/main/java/io/tarantool/pool/PoolEntry.java +++ b/tarantool-pooling/src/main/java/io/tarantool/pool/PoolEntry.java @@ -8,6 +8,7 @@ import java.util.ArrayDeque; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Consumer; import io.micrometer.core.instrument.Counter; @@ -132,9 +133,11 @@ final class PoolEntry { * Flag signaling if connection is available or not. * *

When connection comes to invalidated state or killed, pool entry is locked and connection - * will not be returned to outer client. + * will not be returned to outer client. Connect, close and heartbeat handlers run on different + * threads, so the flag is changed with compare-and-set: only the call that changes it updates the + * count of unavailable connections. */ - private boolean isLocked; + private final AtomicBoolean isLocked = new AtomicBoolean(); /** Count of failed pings occurred in invalidated state. */ private int currentDeathPings; @@ -221,7 +224,6 @@ public PoolEntry( this.gracefulShutdown = gracefulShutdown; this.group = group; this.index = index; - this.isLocked = false; this.reconnectAfter = reconnectAfter; this.tag = group.getTag(); this.timerService = timerService; @@ -251,9 +253,8 @@ public IProtoClient getClient() { *

Also increments count of unavailable clients. */ public void lock() { - if (!isLocked) { + if (isLocked.compareAndSet(false, true)) { this.connectionStatistics.incrementUnavailable(); - isLocked = true; } } @@ -263,10 +264,9 @@ public void lock() { *

Also decrements count of unavailable clients and cancels reconnect task. */ public void unlock() { - if (isLocked) { + if (isLocked.compareAndSet(true, false)) { stopReconnectTask(); this.connectionStatistics.decrementUnavailable(); - isLocked = false; } } @@ -276,7 +276,7 @@ public void unlock() { * @return {@link #isLocked} value. */ public boolean isLocked() { - return isLocked; + return isLocked.get(); } /** Closes client and stops heartbeat and reconnect tasks if started. */ diff --git a/tarantool-pooling/src/test/java/io/tarantool/pool/PoolEntryLockTest.java b/tarantool-pooling/src/test/java/io/tarantool/pool/PoolEntryLockTest.java new file mode 100644 index 00000000..5eac092b --- /dev/null +++ b/tarantool-pooling/src/test/java/io/tarantool/pool/PoolEntryLockTest.java @@ -0,0 +1,113 @@ +/* + * Copyright (c) 2026 VK DIGITAL TECHNOLOGIES LIMITED LIABILITY COMPANY + * All Rights Reserved. + */ + +package io.tarantool.pool; + +import java.net.InetAddress; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.TimeUnit; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import io.netty.bootstrap.Bootstrap; +import io.netty.channel.EventLoopGroup; +import io.netty.channel.MultiThreadIoEventLoopGroup; +import io.netty.channel.nio.NioIoHandler; +import io.netty.channel.socket.nio.NioSocketChannel; +import io.netty.util.HashedWheelTimer; +import io.netty.util.Timer; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import io.tarantool.core.connection.ConnectionFactory; + +/** Lock and unlock of one pool entry called from several threads at once. */ +public class PoolEntryLockTest { + + private static final int THREADS = 8; + private static final int ITERATIONS = 20_000; + + private EventLoopGroup group; + private Timer timer; + + @BeforeEach + public void setUp() { + group = new MultiThreadIoEventLoopGroup(1, NioIoHandler.newFactory()); + timer = new HashedWheelTimer(10, TimeUnit.MILLISECONDS); + } + + @AfterEach + public void tearDown() throws Exception { + timer.stop(); + group.shutdownGracefully().get(); + } + + @Test + public void test_concurrentLockUnlock_shouldKeepUnavailableCountConsistent() throws Exception { + IProtoClientPoolImpl.ConnectionStatistics statistics = + new IProtoClientPoolImpl.ConnectionStatistics(); + PoolEntry entry = createEntry(statistics); + + CyclicBarrier start = new CyclicBarrier(THREADS); + List workers = new ArrayList<>(); + for (int t = 0; t < THREADS; t++) { + Thread worker = + new Thread( + () -> { + try { + start.await(); + } catch (Exception e) { + throw new IllegalStateException(e); + } + for (int i = 0; i < ITERATIONS; i++) { + if (i % 2 == 0) { + entry.lock(); + } else { + entry.unlock(); + } + } + }); + workers.add(worker); + worker.start(); + } + for (Thread worker : workers) { + worker.join(); + } + + // The entry counts as unavailable exactly while it is locked; without an atomic flag two + // threads could both pass the check and change the counter twice. + assertEquals(entry.isLocked() ? 1 : 0, statistics.unavailable()); + } + + private PoolEntry createEntry(IProtoClientPoolImpl.ConnectionStatistics statistics) { + Bootstrap bootstrap = new Bootstrap().group(group).channel(NioSocketChannel.class); + InstanceConnectionGroup instanceGroup = + InstanceConnectionGroup.builder() + .withHost(InetAddress.getLoopbackAddress().getHostAddress()) + .withPort(3301) + .withSize(1) + .withTag("lock") + .build(); + return new PoolEntry( + new ConnectionFactory(bootstrap, timer), + timer, + instanceGroup, + 0, + false, + 1000, + 1000, + null, + null, + statistics, + null, + null, + null, + false, + null, + null); + } +} From 327375657f6aec1d856ae7ab0db3b09491a1274a Mon Sep 17 00:00:00 2001 From: Aleksandr Iuskasov Date: Thu, 22 Oct 2026 21:21:49 +0500 Subject: [PATCH 7/8] docs(pool): describe locked connections Explain when a pool entry is locked (failed connect, close by the server or by a server shutdown, heartbeat invalidation) and unlocked (reconnect, heartbeat activation), that a long request does not lock a connection, and that a request fails with NoAvailableClientsException when all connections are locked. --- CHANGELOG.md | 2 ++ .../pages/client/arch/connection-pool.en.md | 13 +++++++++++++ .../doc-src/pages/client/arch/connection-pool.md | 14 ++++++++++++++ 3 files changed, 29 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index e73d6ec0..fe914748 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -53,6 +53,8 @@ - Added the "Limitations and workarounds" page: no batch delete and no `IN` condition in `crud`, examples of parallel requests instead, and derived-method keywords that are not supported. - Added a table that maps Spring Boot minor lines to `tarantool-spring-data-*` modules. +- Described locked connections of the pool: when a connection is locked and unlocked, and that a + request fails with `NoAvailableClientsException` when all connections are locked. ## [1.7.2] - 2026-09-25 diff --git a/documentation/doc-src/pages/client/arch/connection-pool.en.md b/documentation/doc-src/pages/client/arch/connection-pool.en.md index 9f5d59a3..162150c0 100644 --- a/documentation/doc-src/pages/client/arch/connection-pool.en.md +++ b/documentation/doc-src/pages/client/arch/connection-pool.en.md @@ -28,3 +28,16 @@ requests can be found in the [balancers](balancer.md) section. The actual establishment of the connection in `IProtoClientPool` occurs when the balancer first selects this connection from the pool. + +### Locked connections + +A `PoolEntry` is *locked* when its connection is considered unusable: a connection attempt failed, +the established connection was closed by the server or by a server shutdown, or the +[heartbeat](heartbeat.md) moved the connection to the invalidated state. A locked entry is not +returned to the balancer and is unlocked when the connection is restored (the reconnect succeeds or +the heartbeat reports recovery) or when the entry is removed from a shrinking group. A connection is **not** locked just because a long request is +running on it: many requests are in flight over one iproto connection at the same time. + +The pool counts locked entries. When all entries are locked, the balancer does not wait for one to +be unlocked: the request fails with `NoAvailableClientsException`. A server that is busy so long +that it stops answering pings can lead to this, because the heartbeat invalidates its connections. diff --git a/documentation/doc-src/pages/client/arch/connection-pool.md b/documentation/doc-src/pages/client/arch/connection-pool.md index 40535ec4..0abebc25 100644 --- a/documentation/doc-src/pages/client/arch/connection-pool.md +++ b/documentation/doc-src/pages/client/arch/connection-pool.md @@ -30,3 +30,17 @@ title: Пул соединений Фактическое установление соединения в `IProtoClientPool` происходит, когда балансировщик впервые выбирает это соединение из пула. + +### Заблокированные соединения + +`PoolEntry` *заблокирован*, когда его соединение считается непригодным: попытка подключения +завершилась ошибкой, установленное соединение закрыл сервер или оно закрылось при выключении +сервера, либо [heartbeat](heartbeat.md) перевёл соединение в состояние invalidated. Заблокированная +запись не отдаётся балансировщику и разблокируется, когда соединение восстановлено (переподключение +удалось или heartbeat сообщил о восстановлении) или когда запись удаляется из уменьшаемой группы. Соединение **не** блокируется из-за того, что на нём +выполняется долгий запрос: по одному соединению iproto одновременно идёт много запросов. + +Пул ведёт счёт заблокированных записей. Если заблокированы все записи, балансировщик не ждёт, пока +какая-нибудь освободится: запрос завершается с `NoAvailableClientsException`. К этому может привести +сервер, настолько долго занятый, что не отвечает на пинги: heartbeat признаёт его соединения +недействительными. From 7325a008bb5eab79a9e70a8e1a360dba8871a802 Mon Sep 17 00:00:00 2001 From: Aleksandr Iuskasov Date: Thu, 22 Oct 2026 21:21:49 +0500 Subject: [PATCH 8/8] build(testcontainers): fix module description The description in testcontainers/pom.xml was copied from the connector ("Minimalistic java connector for Tarantool ...") and is shown for the artifact on Maven Central. Describe what the module provides. --- CHANGELOG.md | 5 +++++ testcontainers/pom.xml | 2 +- 2 files changed, 6 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index fe914748..1d9ee0b0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -56,6 +56,11 @@ - Described locked connections of the pool: when a connection is locked and unlocked, and that a request fails with `NoAvailableClientsException` when all connections are locked. +### Testcontainers + +- Fixed the description of the `testcontainers` module in its `pom.xml`: it was copied from the + connector. + ## [1.7.2] - 2026-09-25 ### Client diff --git a/testcontainers/pom.xml b/testcontainers/pom.xml index e6cd35fe..25bbd18b 100644 --- a/testcontainers/pom.xml +++ b/testcontainers/pom.xml @@ -3,7 +3,7 @@ 4.0.0 testcontainers - Minimalistic java connector for Tarantool versions 2.11+ based on Netty framework + Testcontainers for Tarantool: single nodes, vshard and Cartridge clusters, Tarantool DB, TDG, TQE and TCM https://tarantool.io testcontainers