From 16c85642e592bc56caefb5572d336bf7ba9115da Mon Sep 17 00:00:00 2001 From: Alexis Date: Sat, 11 Jul 2026 18:34:12 +0200 Subject: [PATCH 1/4] JAVA-6250 Do not retain the connection-open handler from the channel close listener. The listener registered on the channel closeFuture when a pooled connection opens captured the enclosing OpenChannelFutureListener instance. This kept the open handler, and the whole operation graph reachable from it (callbacks and their buffers), strongly referenced for as long as the pooled channel stayed alive. Capture the enclosing NettyStream in a local variable instead, so the operation graph becomes collectable as soon as the connection opening completes. JAVA-6250 --- .../connection/netty/NettyStream.java | 6 +- .../NettyStreamCloseFutureListenerTest.java | 154 ++++++++++++++++++ 2 files changed, 159 insertions(+), 1 deletion(-) create mode 100644 driver-core/src/test/unit/com/mongodb/internal/connection/netty/NettyStreamCloseFutureListenerTest.java diff --git a/driver-core/src/main/com/mongodb/internal/connection/netty/NettyStream.java b/driver-core/src/main/com/mongodb/internal/connection/netty/NettyStream.java index e480363fc82..d85c011f67c 100644 --- a/driver-core/src/main/com/mongodb/internal/connection/netty/NettyStream.java +++ b/driver-core/src/main/com/mongodb/internal/connection/netty/NettyStream.java @@ -530,7 +530,11 @@ public void operationComplete(final ChannelFuture future) { channelFuture.channel().close(); } else { channel = channelFuture.channel(); - channel.closeFuture().addListener((ChannelFutureListener) future1 -> handleReadResponse(null, new IOException("The connection to the server was closed"))); + // capture only the enclosing stream in the close listener: capturing OpenChannelFutureListener.this + // would pin the open handler (and everything it references) for the lifetime of the pooled channel + NettyStream stream = NettyStream.this; + channel.closeFuture().addListener((ChannelFutureListener) future1 -> + stream.handleReadResponse(null, new IOException("The connection to the server was closed"))); } handler.completed(null); } else { diff --git a/driver-core/src/test/unit/com/mongodb/internal/connection/netty/NettyStreamCloseFutureListenerTest.java b/driver-core/src/test/unit/com/mongodb/internal/connection/netty/NettyStreamCloseFutureListenerTest.java new file mode 100644 index 00000000000..7b256e7b3eb --- /dev/null +++ b/driver-core/src/test/unit/com/mongodb/internal/connection/netty/NettyStreamCloseFutureListenerTest.java @@ -0,0 +1,154 @@ +/* + * Copyright 2008-present MongoDB, Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.mongodb.internal.connection.netty; + +import com.mongodb.ServerAddress; +import com.mongodb.connection.AsyncCompletionHandler; +import com.mongodb.connection.SocketSettings; +import com.mongodb.connection.SslSettings; +import io.netty.buffer.PooledByteBufAllocator; +import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.channel.socket.nio.NioSocketChannel; +import org.bson.ByteBuf; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.lang.ref.WeakReference; +import java.net.InetAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.util.Collections; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import static com.mongodb.ClusterFixture.OPERATION_CONTEXT; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Covers the listener that {@code OpenChannelFutureListener} registers on {@code channel.closeFuture()}, + * from both angles: + * + */ +public class NettyStreamCloseFutureListenerTest { + + private ServerSocket serverSocket; + private NioEventLoopGroup eventLoopGroup; + private NettyStream stream; + + @BeforeEach + public void setUp() throws Exception { + serverSocket = new ServerSocket(0, 1, InetAddress.getLoopbackAddress()); + eventLoopGroup = new NioEventLoopGroup(); + stream = (NettyStream) new NettyStreamFactory( + host -> Collections.singletonList(InetAddress.getLoopbackAddress()), + SocketSettings.builder().connectTimeout(10, TimeUnit.SECONDS).build(), + SslSettings.builder().build(), + eventLoopGroup, NioSocketChannel.class, PooledByteBufAllocator.DEFAULT, null) + .create(new ServerAddress("127.0.0.1", serverSocket.getLocalPort())); + } + + @AfterEach + public void tearDown() throws Exception { + stream.close(); + eventLoopGroup.shutdownGracefully(); + serverSocket.close(); + } + + @Test + @DisplayName("open handler should not remain strongly reachable from the open channel after the open completes") + public void shouldReleaseOpenHandlerAfterOpenCompletesWhileChannelRemainsOpen() throws Exception { + CountDownLatch opened = new CountDownLatch(1); + AsyncCompletionHandler handler = new AsyncCompletionHandler() { + @Override + public void completed(final Void result) { + opened.countDown(); + } + + @Override + public void failed(final Throwable t) { + opened.countDown(); + } + }; + WeakReference> canary = new WeakReference>(handler); + + stream.openAsync(OPERATION_CONTEXT, handler); + assertTrue(opened.await(10, TimeUnit.SECONDS), "open did not complete"); + + handler = null; + for (int i = 0; i < 10 && canary.get() != null; i++) { + System.gc(); + Thread.sleep(100); + } + assertNull(canary.get(), + "the connection-open AsyncCompletionHandler must not stay strongly reachable from the open " + + "channel (closeFuture listener) after the open has completed"); + } + + @Test + @DisplayName("pending read should be failed when the channel is closed") + public void shouldFailPendingReadWhenChannelIsClosed() throws Exception { + AtomicReference acceptedSocket = new AtomicReference<>(); + Thread acceptor = new Thread(() -> { + try { + acceptedSocket.set(serverSocket.accept()); + } catch (IOException ignored) { + // the assertions below fail if nothing was accepted + } + }); + acceptor.start(); + + stream.open(OPERATION_CONTEXT); + acceptor.join(TimeUnit.SECONDS.toMillis(10)); + assertNotNull(acceptedSocket.get(), "the server never accepted the connection"); + + CountDownLatch readCompleted = new CountDownLatch(1); + AtomicReference readFailure = new AtomicReference<>(); + stream.readAsync(4, OPERATION_CONTEXT, new AsyncCompletionHandler() { + @Override + public void completed(final ByteBuf result) { + readCompleted.countDown(); + } + + @Override + public void failed(final Throwable t) { + readFailure.set(t); + readCompleted.countDown(); + } + }); + + acceptedSocket.get().close(); + + assertTrue(readCompleted.await(10, TimeUnit.SECONDS), "the pending read never completed"); + Throwable failure = readFailure.get(); + assertNotNull(failure, "the pending read completed successfully instead of failing"); + assertInstanceOf(IOException.class, failure); + assertEquals("The connection to the server was closed", failure.getMessage()); + } +} From d074ea43ae9dcc0fab3ccce72c7834bf2868c5ba Mon Sep 17 00:00:00 2001 From: Ross Lawley Date: Mon, 20 Jul 2026 10:19:08 +0100 Subject: [PATCH 2/4] Extract channel close listener into static CloseChannelFutureListener The close listener now captures only the NettyStream, so the one-shot connection-open handler can no longer be pinned for the channel lifetime. Replaces the local-variable capture idiom with a structural guarantee. JAVA-6250 --- .../connection/netty/NettyStream.java | 21 ++++++++++++++----- 1 file changed, 16 insertions(+), 5 deletions(-) diff --git a/driver-core/src/main/com/mongodb/internal/connection/netty/NettyStream.java b/driver-core/src/main/com/mongodb/internal/connection/netty/NettyStream.java index d85c011f67c..d41402e0ad7 100644 --- a/driver-core/src/main/com/mongodb/internal/connection/netty/NettyStream.java +++ b/driver-core/src/main/com/mongodb/internal/connection/netty/NettyStream.java @@ -530,11 +530,9 @@ public void operationComplete(final ChannelFuture future) { channelFuture.channel().close(); } else { channel = channelFuture.channel(); - // capture only the enclosing stream in the close listener: capturing OpenChannelFutureListener.this - // would pin the open handler (and everything it references) for the lifetime of the pooled channel - NettyStream stream = NettyStream.this; - channel.closeFuture().addListener((ChannelFutureListener) future1 -> - stream.handleReadResponse(null, new IOException("The connection to the server was closed"))); + // CloseChannelFutureListener captures only the NettyStream, never OpenChannelFutureListener.this, + // so the one-shot connection-open handler is not pinned for the lifetime of the pooled channel (JAVA-6250) + channel.closeFuture().addListener(new CloseChannelFutureListener(NettyStream.this)); } handler.completed(null); } else { @@ -550,6 +548,19 @@ public void operationComplete(final ChannelFuture future) { } } + private static final class CloseChannelFutureListener implements ChannelFutureListener { + private final NettyStream stream; + + CloseChannelFutureListener(final NettyStream stream) { + this.stream = stream; + } + + @Override + public void operationComplete(final ChannelFuture future) { + stream.handleReadResponse(null, new IOException("The connection to the server was closed")); + } + } + private static void cancel(@Nullable final Future f) { if (f != null) { f.cancel(false); From 69cae2565ff22320bed88b9d9535ec3d90c3061e Mon Sep 17 00:00:00 2001 From: Ross Lawley Date: Mon, 20 Jul 2026 11:08:48 +0100 Subject: [PATCH 3/4] Convert OpenChannelFutureListener to a static nested class Explicit NettyStream dependency instead of the implicit enclosing instance, removing the this$0 capture footgun that caused JAVA-6250. No behavioral change. JAVA-6250 --- .../connection/netty/NettyStream.java | 22 ++++++++++--------- 1 file changed, 12 insertions(+), 10 deletions(-) diff --git a/driver-core/src/main/com/mongodb/internal/connection/netty/NettyStream.java b/driver-core/src/main/com/mongodb/internal/connection/netty/NettyStream.java index d41402e0ad7..34e1308cd47 100644 --- a/driver-core/src/main/com/mongodb/internal/connection/netty/NettyStream.java +++ b/driver-core/src/main/com/mongodb/internal/connection/netty/NettyStream.java @@ -221,7 +221,7 @@ public void initChannel(final SocketChannel ch) { } }); ChannelFuture channelFuture = bootstrap.connect(nextAddress); - channelFuture.addListener(new OpenChannelFutureListener(operationContext, socketAddressQueue, channelFuture, handler)); + channelFuture.addListener(new OpenChannelFutureListener(this, operationContext, socketAddressQueue, channelFuture, handler)); } } @@ -507,15 +507,17 @@ public T get() throws IOException { } } - private class OpenChannelFutureListener implements ChannelFutureListener { + private static final class OpenChannelFutureListener implements ChannelFutureListener { + private final NettyStream stream; private final Queue socketAddressQueue; private final ChannelFuture channelFuture; private final AsyncCompletionHandler handler; private final OperationContext operationContext; - OpenChannelFutureListener(final OperationContext operationContext, + OpenChannelFutureListener(final NettyStream stream, final OperationContext operationContext, final Queue socketAddressQueue, final ChannelFuture channelFuture, final AsyncCompletionHandler handler) { + this.stream = stream; this.operationContext = operationContext; this.socketAddressQueue = socketAddressQueue; this.channelFuture = channelFuture; @@ -524,24 +526,24 @@ private class OpenChannelFutureListener implements ChannelFutureListener { @Override public void operationComplete(final ChannelFuture future) { - withLock(lock, () -> { + withLock(stream.lock, () -> { if (future.isSuccess()) { - if (isClosed) { + if (stream.isClosed) { channelFuture.channel().close(); } else { - channel = channelFuture.channel(); + stream.channel = channelFuture.channel(); // CloseChannelFutureListener captures only the NettyStream, never OpenChannelFutureListener.this, // so the one-shot connection-open handler is not pinned for the lifetime of the pooled channel (JAVA-6250) - channel.closeFuture().addListener(new CloseChannelFutureListener(NettyStream.this)); + stream.channel.closeFuture().addListener(new CloseChannelFutureListener(stream)); } handler.completed(null); } else { - if (isClosed) { + if (stream.isClosed) { handler.completed(null); } else if (socketAddressQueue.isEmpty()) { - handler.failed(new MongoSocketOpenException("Exception opening socket", getAddress(), future.cause())); + handler.failed(new MongoSocketOpenException("Exception opening socket", stream.getAddress(), future.cause())); } else { - initializeChannel(operationContext, handler, socketAddressQueue); + stream.initializeChannel(operationContext, handler, socketAddressQueue); } } }); From a209163655f57f153df7891dc211704f577de34a Mon Sep 17 00:00:00 2001 From: Ross Lawley Date: Mon, 20 Jul 2026 11:08:49 +0100 Subject: [PATCH 4/4] Improve readability of NettyStreamCloseFutureListenerTest Extract the GC reachability probe into a named assertRefUnreachable helper that documents the WeakReference-as-reachability-probe idiom, extract the accept plumbing into openAndAcceptConnection, and add explanatory comments tying the footprint assertion back to JAVA-6250. Test-only, no behaviour change. JAVA-6250 --- .../NettyStreamCloseFutureListenerTest.java | 73 +++++++++++++------ 1 file changed, 52 insertions(+), 21 deletions(-) diff --git a/driver-core/src/test/unit/com/mongodb/internal/connection/netty/NettyStreamCloseFutureListenerTest.java b/driver-core/src/test/unit/com/mongodb/internal/connection/netty/NettyStreamCloseFutureListenerTest.java index 7b256e7b3eb..2287b478938 100644 --- a/driver-core/src/test/unit/com/mongodb/internal/connection/netty/NettyStreamCloseFutureListenerTest.java +++ b/driver-core/src/test/unit/com/mongodb/internal/connection/netty/NettyStreamCloseFutureListenerTest.java @@ -77,13 +77,14 @@ public void setUp() throws Exception { @AfterEach public void tearDown() throws Exception { stream.close(); - eventLoopGroup.shutdownGracefully(); + eventLoopGroup.shutdownGracefully().syncUninterruptibly(); serverSocket.close(); } @Test @DisplayName("open handler should not remain strongly reachable from the open channel after the open completes") public void shouldReleaseOpenHandlerAfterOpenCompletesWhileChannelRemainsOpen() throws Exception { + // Open a connection with a handler, and keep only a WeakReference to that handler CountDownLatch opened = new CountDownLatch(1); AsyncCompletionHandler handler = new AsyncCompletionHandler() { @Override @@ -96,17 +97,19 @@ public void failed(final Throwable t) { opened.countDown(); } }; - WeakReference> canary = new WeakReference>(handler); + WeakReference> canary = new WeakReference<>(handler); stream.openAsync(OPERATION_CONTEXT, handler); assertTrue(opened.await(10, TimeUnit.SECONDS), "open did not complete"); + // Nullify the test's own reference so the driver is the only thing that could still retain the handler. + // Note: the channel is deliberately left open (never closed) for the rest of the test, so a positive + // result proves the handler is released while the channel is alive. handler = null; - for (int i = 0; i < 10 && canary.get() != null; i++) { - System.gc(); - Thread.sleep(100); - } - assertNull(canary.get(), + + // The channel's closeFuture listener must not keep the one-shot open handler alive, + // so with no strong reference remaining the handler must be garbage-collectable. + assertRefUnreachable(canary, "the connection-open AsyncCompletionHandler must not stay strongly reachable from the open " + "channel (closeFuture listener) after the open has completed"); } @@ -114,20 +117,9 @@ public void failed(final Throwable t) { @Test @DisplayName("pending read should be failed when the channel is closed") public void shouldFailPendingReadWhenChannelIsClosed() throws Exception { - AtomicReference acceptedSocket = new AtomicReference<>(); - Thread acceptor = new Thread(() -> { - try { - acceptedSocket.set(serverSocket.accept()); - } catch (IOException ignored) { - // the assertions below fail if nothing was accepted - } - }); - acceptor.start(); - - stream.open(OPERATION_CONTEXT); - acceptor.join(TimeUnit.SECONDS.toMillis(10)); - assertNotNull(acceptedSocket.get(), "the server never accepted the connection"); + Socket acceptedSocket = openAndAcceptConnection(); + // Create a read that 127.0.0.1 will never satisfy, then close the connection from the server side CountDownLatch readCompleted = new CountDownLatch(1); AtomicReference readFailure = new AtomicReference<>(); stream.readAsync(4, OPERATION_CONTEXT, new AsyncCompletionHandler() { @@ -143,12 +135,51 @@ public void failed(final Throwable t) { } }); - acceptedSocket.get().close(); + acceptedSocket.close(); + // Assert the closeFuture listener fires and fails the pending read with the expected IOException assertTrue(readCompleted.await(10, TimeUnit.SECONDS), "the pending read never completed"); Throwable failure = readFailure.get(); assertNotNull(failure, "the pending read completed successfully instead of failing"); assertInstanceOf(IOException.class, failure); assertEquals("The connection to the server was closed", failure.getMessage()); } + + /** + * Asserts that the referent of {@code ref} is no longer strongly reachable, i.e. no strong reference to it + * remains and it is therefore garbage-collectable. + * A {@link WeakReference} is a reachability probe: the garbage collector clears it as soon as its referent is + * no longer strongly reachable. {@link System#gc()} is only a hint, so we nudge it a few times and give the + * collector a moment before checking - if the referent is genuinely unreachable the reference clears quickly. + */ + @SuppressWarnings("BusyWait") + private static void assertRefUnreachable(final WeakReference ref, final String message) throws InterruptedException { + long deadlineNanos = System.nanoTime() + TimeUnit.SECONDS.toNanos(10); + while (ref.get() != null && System.nanoTime() < deadlineNanos) { + System.gc(); + Thread.sleep(100); + } + assertNull(ref.get(), message); + } + + /** + * Opens {@link #stream} against the local {@link #serverSocket} and returns the server side of the accepted + * connection, so the test can later close it to simulate the server dropping the connection. + */ + private Socket openAndAcceptConnection() throws Exception { + AtomicReference acceptedSocket = new AtomicReference<>(); + Thread acceptor = new Thread(() -> { + try { + acceptedSocket.set(serverSocket.accept()); + } catch (IOException ignored) { + // the assertion below fails if nothing was accepted + } + }); + acceptor.start(); + + stream.open(OPERATION_CONTEXT); + acceptor.join(TimeUnit.SECONDS.toMillis(10)); + assertNotNull(acceptedSocket.get(), "the server never accepted the connection"); + return acceptedSocket.get(); + } }