From e5f711b4b4444a736a98d3bb248b1d13500c6fd5 Mon Sep 17 00:00:00 2001 From: scottf Date: Wed, 29 Jul 2026 12:06:40 -0400 Subject: [PATCH] Close a dead passive during reconnect, freeing up resources. --- .../io/nats/client/impl/ApConnection.java | 22 ++++++-- .../java/io/nats/client/impl/ApTests.java | 55 +++++++++++++++++++ 2 files changed, 71 insertions(+), 6 deletions(-) diff --git a/src/main/java/io/nats/client/impl/ApConnection.java b/src/main/java/io/nats/client/impl/ApConnection.java index 2d8e19c..239a01c 100644 --- a/src/main/java/io/nats/client/impl/ApConnection.java +++ b/src/main/java/io/nats/client/impl/ApConnection.java @@ -234,13 +234,23 @@ protected void reconnectImplConnect() throws InterruptedException { } if (!passiveConnection.isConnected()) { - // The passive is also down (e.g. the whole cluster is restarting), so there's no live - // socket to promote. Drop it and let the base logic reconnect the active through the - // server pool. Once the active is back, RE-ARM the passive: a warm standby must ALWAYS - // be re-established, or the first full-cluster outage would permanently leave us with no - // passive. Best-effort (reArmPassive) so a passive that can't connect yet never aborts - // the active's reconnect; the next active reconnect will try again. + // The passive is also down (e.g. the whole cluster is restarting), so there's no live socket to + // promote. CLOSE it - don't just drop the reference - before letting the active reconnect. The + // passive is its own NatsConnection with maxReconnects(-1), so its independent reconnect thread + // would otherwise (a) compete with the active's reconnect, both hammering the recovering brokers, + // and (b) leak: newPassive() below re-arms from a now-null field, so nothing would ever close this + // dead passive - its thread would run forever and reconnect as a phantom on recovery. + // Null the field FIRST so a re-entrant reconnectImplConnect (the passiveConnection == null guard + // above) short-circuits if handleCommunicationIssue fires again during the close. + // Executor-safe here, unlike the socket-steal path below: no adopted reader runs on the passive's + // executor, and Options rebuilds shut-down internal executors on next use, so the re-armed passive + // still gets a working pool. + // Once the active is back, RE-ARM the passive: a warm standby must ALWAYS be re-established, or the + // first full-cluster outage would permanently leave us with no passive. Best-effort (reArmPassive) + // so a passive that can't connect yet never aborts the active's reconnect; the next reconnect retries. + NatsConnection deadPassive = passiveConnection; passiveConnection = null; + deadPassive.close(); super.reconnectImplConnect(); if (isConnected()) { activeConnectSucceeded(); diff --git a/src/test/java/io/nats/client/impl/ApTests.java b/src/test/java/io/nats/client/impl/ApTests.java index 5604764..717e8fe 100644 --- a/src/test/java/io/nats/client/impl/ApTests.java +++ b/src/test/java/io/nats/client/impl/ApTests.java @@ -8,11 +8,13 @@ import io.nats.client.Options; import io.nats.client.support.Listener; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; import java.io.IOException; import java.util.ArrayList; import java.util.List; import java.util.concurrent.CompletableFuture; +import java.util.function.BooleanSupplier; import java.util.logging.Level; import static org.junit.jupiter.api.Assertions.*; @@ -367,4 +369,57 @@ public void testSwitchToPassiveClearsOutstandingPings() throws Exception { } } } + + private static boolean waitFor(BooleanSupplier cond, long millis) throws InterruptedException { + long end = System.currentTimeMillis() + millis; + while (System.currentTimeMillis() < end) { + if (cond.getAsBoolean()) { + return true; + } + Thread.sleep(25); + } + return cond.getAsBoolean(); + } + + // Issue #24: when the passive is down at reconnect time (whole-cluster path), reconnectImplConnect must + // CLOSE the dead passive, not just drop the reference - otherwise its maxReconnects(-1) reconnect thread + // runs on as an orphan (competing load + a leak that reconnects as a phantom on recovery). The close + // happens BEFORE the active's own reconnect loop, so we can observe it while the cluster is still down. + @Test + @Timeout(90) + public void testWholeClusterReconnectClosesDeadPassive() throws Exception { + NatsServerRunner server1 = new NatsServerRunner(); + NatsServerRunner server2 = new NatsServerRunner(); + try { + // active = server1, passive = server2 (distinct), no randomize so it's deterministic + Options.Builder builder = new Options.Builder(getOptions(server1, server2)).noRandomize(); + OptionsHelper helper = new OptionsHelper(builder); + try (ApConnection apc = ApConnection.connect(helper.apOptions)) { + helper.validateConnected(); + NatsConnection deadPassive = apc.passiveConnection; + assertNotNull(deadPassive); + assertEquals(Connection.Status.CONNECTED, deadPassive.getStatus()); + assertNotEquals(apc.getServerInfo().getServerId(), deadPassive.getServerInfo().getServerId()); + + // 1) take the passive's server down; with only server1 (the active, which the pool skips) left, + // the passive has nowhere distinct to reconnect and stays not-connected. + server2.close(); + assertTrue(waitFor(() -> !deadPassive.isConnected(), 15000), + "passive should drop when its server is gone"); + + // 2) take the active's server down too (whole cluster now down). The active's own reconnect + // runs on a background thread and hits the whole-cluster branch with a non-connected passive. + server1.close(); + + // THE FIX: the whole-cluster branch CLOSES the dead passive (before it even starts looping to + // reconnect the active), rather than orphaning its reconnect thread. + assertTrue(waitFor(() -> deadPassive.getStatus() == Connection.Status.CLOSED, 20000), + "the dead passive must be CLOSED after the whole-cluster reconnect, not left running"); + } + } + finally { + try { server1.close(); } catch (Exception ignore) { } + try { server2.close(); } catch (Exception ignore) { } + } + } }