From a2e00767159e5977a5bc5028e6bf609be6039a95 Mon Sep 17 00:00:00 2001 From: Aleksandr Iuskasov Date: Wed, 21 Oct 2026 21:38:14 +0500 Subject: [PATCH 1/5] 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. 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 a 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 | 49 +++++- .../ConnectionToTarantoolTest.java | 62 +++++++- .../pool/unit/ReconnectLoopTest.java | 147 ++++++++++++++++++ 4 files changed, 253 insertions(+), 11 deletions(-) create mode 100644 tarantool-pooling/src/test/java/io/tarantool/pool/unit/ReconnectLoopTest.java diff --git a/CHANGELOG.md b/CHANGELOG.md index 16d5967f..d25ff3fb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -68,6 +68,12 @@ `VHANDSHAKE_NOT_COMPLETE`. The check now makes a synchronous `callrw` to every replicaset master, which reopens a router connection that vshard closed after a failed handshake and waits for the new one, requires every master to be `available`, and logs the replica and bucket states. +- 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`, 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..a05167fb 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,8 @@ 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_CONNECTION_TIMEOUT = "Connection timeout"; 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 +122,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,6 +198,7 @@ public synchronized CompletableFuture connect(InetSocketAddress addres log.debug("{} is connecting to {}", this, address); CompletableFuture promise = new CompletableFuture<>(); + greetingPromise = promise; scheduleGreetingTimeout(promise, timeoutMs); connectPromise = promise.thenApply( @@ -358,6 +367,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(); @@ -421,18 +439,35 @@ private void scheduleGreetingTimeout(CompletableFuture promise, long t 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(ERR_CONNECTION_TIMEOUT))) { + 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/integration/ConnectionToTarantoolTest.java b/tarantool-core/src/test/java/io/tarantool/core/integration/ConnectionToTarantoolTest.java index 77ea1ee4..b8391712 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 @@ -8,7 +8,10 @@ import java.io.IOException; import java.lang.reflect.Field; import java.net.ConnectException; +import java.net.InetAddress; import java.net.InetSocketAddress; +import java.net.ServerSocket; +import java.net.Socket; import java.net.SocketException; import java.nio.channels.ClosedChannelException; import java.time.Duration; @@ -397,18 +400,69 @@ public void testConnectByMultipleThreads() throws InterruptedException { assertEquals(CONCURRENT_THREADS_COUNT, success); } + /* + * Server accepts the connection but does not send the greeting. The connect future fails only + * with TimeoutException, and no CLOSE_BY_REMOTE event is fired: the client closes the channel + * itself. + */ @Test @DisabledIfEnvironmentVariable(named = "TARANTOOL_VERSION", matches = "3.*") public void testConnectWithGreetingTimeout() throws InterruptedException { - Connection client = factory.create(); + AtomicReference closedByRemote = new AtomicReference<>(); + Connection client = + factory + .create() + .onClose(ConnectionCloseEvent.CLOSE_BY_REMOTE, (c, ex) -> closedByRemote.set(ex)); InetSocketAddress address = new InetSocketAddress(tt.getHost(), tt.getMappedPort(3306)); CompletableFuture future = client.connect(address, 1_000); Exception ex = assertThrows(CompletionException.class, future::join); Throwable cause = ex.getCause(); - assertTrue( - cause instanceof TimeoutException || cause instanceof ConnectionClosedException, - "Exception should be either TimeoutException or ConnectionClosedException"); + assertEquals(TimeoutException.class, cause.getClass()); assertEquals("Connection timeout", cause.getMessage()); + // Without the fix the close event follows the timeout within milliseconds. + Thread.sleep(500); + assertNull(closedByRemote.get(), "greeting timeout is reported as closed by server"); + } + + /* + * The server closes the first attempt before the greeting; the second attempt waits for the + * greeting of a silent server. The greeting timer of the first attempt must not close the second + * one: it fails with TimeoutException after its own timeout. + */ + @Test + @DisabledIfEnvironmentVariable(named = "TARANTOOL_VERSION", matches = "3.*") + public void testGreetingTimerOfClosedAttemptDoesNotCloseNextAttempt() throws Exception { + long timeoutMs = 1_000; + long pauseMs = 500; + Connection client = factory.create(); + try (ServerSocket closingServer = new ServerSocket(0, 50, InetAddress.getLoopbackAddress())) { + Thread closer = + new Thread( + () -> { + try (Socket socket = closingServer.accept()) { + // closed by try-with-resources before the greeting + } catch (IOException ignored) { + // the test fails on its own assertions + } + }); + closer.setDaemon(true); + closer.start(); + InetSocketAddress closing = + new InetSocketAddress(closingServer.getInetAddress(), closingServer.getLocalPort()); + assertThrows(CompletionException.class, client.connect(closing, timeoutMs)::join); + } + Thread.sleep(pauseMs); + + InetSocketAddress silent = new InetSocketAddress(tt.getHost(), tt.getMappedPort(3306)); + long start = System.nanoTime(); + CompletableFuture second = client.connect(silent, timeoutMs); + Throwable cause = assertThrows(CompletionException.class, second::join).getCause(); + long elapsedMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); + + // Without the fix the timer of the first attempt closes the second one about + // timeoutMs - pauseMs after it starts, and it fails as closed by server. + assertEquals(TimeoutException.class, cause.getClass(), cause.toString()); + assertTrue(elapsedMs >= timeoutMs - 100, "second attempt failed after " + elapsedMs + " ms"); } @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 cb783f2a638686425b123a0c8e715e463d554a9d Mon Sep 17 00:00:00 2001 From: Aleksandr Iuskasov Date: Thu, 22 Oct 2026 21:36:56 +0500 Subject: [PATCH 2/5] 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 d25ff3fb..bbcf8280 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -74,6 +74,10 @@ 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 a05167fb..3bc2debc 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 @@ -126,6 +126,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; /** @@ -199,6 +205,7 @@ public synchronized CompletableFuture connect(InetSocketAddress addres CompletableFuture promise = new CompletableFuture<>(); greetingPromise = promise; + established = false; scheduleGreetingTimeout(promise, timeoutMs); connectPromise = promise.thenApply( @@ -206,6 +213,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; }); @@ -395,13 +403,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 b0a1485d09ace39809d964af6b1375ff70ea2390 Mon Sep 17 00:00:00 2001 From: Aleksandr Iuskasov Date: Wed, 21 Oct 2026 21:39:35 +0500 Subject: [PATCH 3/5] 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 bbcf8280..8c1dc5a2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -78,6 +78,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 afde2a67df4e74acb20ce2aea2e672db6bfb47fe Mon Sep 17 00:00:00 2001 From: Aleksandr Iuskasov Date: Thu, 22 Oct 2026 21:17:03 +0500 Subject: [PATCH 4/5] 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 8c1dc5a2..2abf1607 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -82,6 +82,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 be673fbabde3839988ffa94970c0830e07685624 Mon Sep 17 00:00:00 2001 From: Aleksandr Iuskasov Date: Thu, 22 Oct 2026 21:17:33 +0500 Subject: [PATCH 5/5] 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 2abf1607..8ecc5ca0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -84,6 +84,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); + } +}