Fix ByteBuf resource leaks in connection handling

.evergreen/.evg.yml

@@ -2430,18 +2430,27 @@ buildvariants:
       - name: "socket-test-task"
 
   - matrix_name: "tests-netty"
-    matrix_spec: { auth: "noauth", ssl: "*", jdk: "jdk8", version: [ "7.0" ], topology: "replicaset", os: "linux",
+    matrix_spec: { auth: "noauth", ssl: "*", jdk: "jdk8", version: [ "8.0" ], topology: "replicaset", os: "linux",
                    async-transport: "netty" }
     display_name: "Netty: ${version} ${topology} ${ssl} ${auth} ${jdk} ${os} "
     tags: [ "tests-netty-variant" ]
+    tasks:
+      - name: "test-core-task"
+      - name: "test-reactive-task"
+
+  - matrix_name: "tests-netty-compression"
+    matrix_spec: { compressor: "snappy", auth: "noauth", ssl: "nossl", jdk: "jdk21", version: [ "8.0" ], topology: "replicaset",
+                   os: "linux", async-transport: "netty" }
+    display_name: "Netty with Compression '${compressor}': ${version} ${topology} ${ssl} ${auth} ${jdk} ${os} "
+    tags: [ "tests-netty-variant" ]
     tasks:
       - name: "test-reactive-task"
       - name: "test-core-task"
 
   - matrix_name: "tests-netty-ssl-provider"
-    matrix_spec: { auth: "auth", ssl: "ssl", jdk: "jdk8", version: [ "7.0" ], topology: "replicaset", os: "linux",
+    matrix_spec: { auth: "auth", ssl: "ssl", jdk: "jdk8", version: [ "8.0" ], topology: "replicaset", os: "linux",
                    async-transport: "netty", netty-ssl-provider: "*" }
-    display_name: "Netty SSL provider: ${version} ${topology} ${ssl} SslProvider.${netty-ssl-provider} ${auth} ${jdk} ${os} "
+    display_name: "Netty with SSL provider '${netty-ssl-provider}': ${version} ${topology} ${ssl} ${auth} ${jdk} ${os} "
     tags: [ "tests-netty-variant" ]
     tasks:
       - name: "test-reactive-task"

.evergreen/run-tests.sh

@@ -67,14 +67,17 @@ provision_ssl () {
 }
 
 provision_multi_mongos_uri_for_ssl () {
-  # Arguments for auth + SSL
-  if [ "$AUTH" != "noauth" ] || [ "$TOPOLOGY" == "replica_set" ]; then
-    export MONGODB_URI="${MONGODB_URI}&ssl=true&sslInvalidHostNameAllowed=true"
-    export MULTI_MONGOS_URI="${MULTI_MONGOS_URI}&ssl=true&sslInvalidHostNameAllowed=true"
-  else
-    export MONGODB_URI="${MONGODB_URI}/?ssl=true&sslInvalidHostNameAllowed=true"
-    export MULTI_MONGOS_URI="${MULTI_MONGOS_URI}/?ssl=true&sslInvalidHostNameAllowed=true"
-  fi
+     if [[ "$MONGODB_URI" == *"?"* ]]; then
+       export MONGODB_URI="${MONGODB_URI}&ssl=true&sslInvalidHostNameAllowed=true"
+     else
+       export MONGODB_URI="${MONGODB_URI}/?ssl=true&sslInvalidHostNameAllowed=true"
+     fi
+
+     if [[ "$MULTI_MONGOS_URI" == *"?"* ]]; then
+       export MULTI_MONGOS_URI="${MULTI_MONGOS_URI}&ssl=true&sslInvalidHostNameAllowed=true"
+     else
+       export MULTI_MONGOS_URI="${MULTI_MONGOS_URI}/?ssl=true&sslInvalidHostNameAllowed=true"
+     fi
 }
 
 ############################################
@@ -136,6 +139,8 @@ echo "Running tests with Java ${JAVA_VERSION}"
           --stacktrace --info --continue ${TESTS} | tee -a logs.txt
 
 if grep -q 'LEAK:' logs.txt ; then
-    echo "Netty Leak detected, please inspect build log"
+    echo "==============================================="
+    echo " Netty Leak detected, please inspect build log "
+    echo "==============================================="
     exit 1
 fi

bson/src/main/org/bson/BsonDocument.java

@@ -921,10 +921,12 @@ private static class SerializationProxy implements Serializable {
             new BsonDocumentCodec().encode(new BsonBinaryWriter(buffer), document, EncoderContext.builder().build());
             this.bytes = new byte[buffer.size()];
             int curPos = 0;
-            for (ByteBuf cur : buffer.getByteBuffers()) {
+            List<ByteBuf> byteBuffers = buffer.getByteBuffers();
+            for (ByteBuf cur : byteBuffers) {
                 System.arraycopy(cur.array(), cur.position(), bytes, curPos, cur.limit());
                 curPos += cur.position();
             }
+            byteBuffers.forEach(ByteBuf::release);
         }
 
         private Object readResolve() {

config/spotbugs/exclude.xml

@@ -290,4 +290,10 @@
         <Bug pattern="NM_CLASS_NAMING_CONVENTION"/>
     </Match>
 
+    <!-- DefaultServerMonitor -->
+    <Match>
+        <class name="com.mongodb.internal.connection.DefaultServerMonitor" />
+        <Bug pattern="RCN_REDUNDANT_NULLCHECK_OF_NONNULL_VALUE" />
+    </Match>
+
 </FindBugsFilter>

driver-core/src/main/com/mongodb/internal/connection/ByteBufBsonArray.java

@@ -23,6 +23,7 @@
 import org.bson.ByteBuf;
 import org.bson.io.ByteBufferBsonInput;
 
+import java.io.Closeable;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Iterator;
@@ -33,20 +34,31 @@
 
 import static com.mongodb.internal.connection.ByteBufBsonHelper.readBsonValue;
 
-final class ByteBufBsonArray extends BsonArray {
+final class ByteBufBsonArray extends BsonArray implements Closeable {
     private final ByteBuf byteBuf;
 
+    /**
+     * List of resources that need to be closed when this array is closed.
+     * Tracks the main ByteBuf and iterator duplicates. Iterator buffers are automatically
+     * removed and released when iteration completes normally to prevent memory accumulation.
+     */
+    private final List<Closeable> trackedResources = new ArrayList<>();
+    private boolean closed;
+
     ByteBufBsonArray(final ByteBuf byteBuf) {
         this.byteBuf = byteBuf;
+        trackedResources.add(byteBuf::release);
     }
 
     @Override
     public Iterator<BsonValue> iterator() {
+        ensureOpen();
         return new ByteBufBsonArrayIterator();
     }
 
     @Override
     public List<BsonValue> getValues() {
+        ensureOpen();
         List<BsonValue> values = new ArrayList<>();
         for (BsonValue cur: this) {
             //noinspection UseBulkOperation
@@ -59,6 +71,7 @@ public List<BsonValue> getValues() {
 
     @Override
     public int size() {
+        ensureOpen();
         int size = 0;
         for (BsonValue ignored : this) {
             size++;
@@ -68,11 +81,13 @@ public int size() {
 
     @Override
     public boolean isEmpty() {
+        ensureOpen();
         return !iterator().hasNext();
     }
 
     @Override
     public boolean equals(final Object o) {
+        ensureOpen();
         if (o == this) {
             return true;
         }
@@ -91,6 +106,7 @@ public boolean equals(final Object o) {
 
     @Override
     public int hashCode() {
+        ensureOpen();
         int hashCode = 1;
         for (BsonValue cur : this) {
             hashCode = 31 * hashCode + (cur == null ? 0 : cur.hashCode());
@@ -100,6 +116,7 @@ public int hashCode() {
 
     @Override
     public boolean contains(final Object o) {
+        ensureOpen();
         for (BsonValue cur : this) {
             if (Objects.equals(o, cur)) {
                 return true;
@@ -111,6 +128,7 @@ public boolean contains(final Object o) {
 
     @Override
     public Object[] toArray() {
+        ensureOpen();
         Object[] retVal = new Object[size()];
         Iterator<BsonValue> it = iterator();
         for (int i = 0; i < retVal.length; i++) {
@@ -122,6 +140,7 @@ public Object[] toArray() {
     @Override
     @SuppressWarnings("unchecked")
     public <T> T[] toArray(final T[] a) {
+        ensureOpen();
         int size = size();
         T[] retVal = a.length >= size ? a : (T[]) java.lang.reflect.Array.newInstance(a.getClass().getComponentType(), size);
         Iterator<BsonValue> it = iterator();
@@ -133,6 +152,7 @@ public <T> T[] toArray(final T[] a) {
 
     @Override
     public boolean containsAll(final Collection<?> c) {
+        ensureOpen();
         for (Object e : c) {
             if (!contains(e)) {
                 return false;
@@ -143,6 +163,7 @@ public boolean containsAll(final Collection<?> c) {
 
     @Override
     public BsonValue get(final int index) {
+        ensureOpen();
         if (index < 0) {
             throw new IndexOutOfBoundsException("Index out of range: " + index);
         }
@@ -159,6 +180,7 @@ public BsonValue get(final int index) {
 
     @Override
     public int indexOf(final Object o) {
+        ensureOpen();
         int i = 0;
         for (BsonValue cur : this) {
             if (Objects.equals(o, cur)) {
@@ -172,6 +194,7 @@ public int indexOf(final Object o) {
 
     @Override
     public int lastIndexOf(final Object o) {
+        ensureOpen();
         ListIterator<BsonValue> listIterator = listIterator(size());
         while (listIterator.hasPrevious()) {
             if (Objects.equals(o, listIterator.previous())) {
@@ -183,17 +206,20 @@ public int lastIndexOf(final Object o) {
 
     @Override
     public ListIterator<BsonValue> listIterator() {
+        ensureOpen();
         return listIterator(0);
     }
 
     @Override
     public ListIterator<BsonValue> listIterator(final int index) {
+        ensureOpen();
         // Not the most efficient way to do this, but unlikely anyone will notice in practice
         return new ArrayList<>(this).listIterator(index);
     }
 
     @Override
     public List<BsonValue> subList(final int fromIndex, final int toIndex) {
+        ensureOpen();
         if (fromIndex < 0) {
             throw new IndexOutOfBoundsException("fromIndex = " + fromIndex);
         }
@@ -234,6 +260,7 @@ public boolean addAll(final Collection<? extends BsonValue> c) {
 
     @Override
     public boolean addAll(final int index, final Collection<? extends BsonValue> c) {
+        ensureOpen();
         throw new UnsupportedOperationException(READ_ONLY_MESSAGE);
     }
 
@@ -267,11 +294,43 @@ public BsonValue remove(final int index) {
         throw new UnsupportedOperationException(READ_ONLY_MESSAGE);
     }
 
+    @Override
+    public void close(){
+        if (!closed) {
+            for (Closeable closeable : trackedResources) {
+                try {
+                    closeable.close();
+                } catch (Exception e) {
+                    // Log and continue closing other resources
+                }
+            }
+            trackedResources.clear();
+            closed = true;
+        }
+    }
+
+    private void ensureOpen() {
+        if (closed) {
+            throw new IllegalStateException("The BsonArray resources have been released.");
+        }
+    }
+
     private class ByteBufBsonArrayIterator implements Iterator<BsonValue> {
-        private final ByteBuf duplicatedByteBuf = byteBuf.duplicate();
-        private final BsonBinaryReader bsonReader;
+        private ByteBuf duplicatedByteBuf;
+        private BsonBinaryReader bsonReader;
+        private Closeable resourceHandle;
+        private boolean finished;
 
         {
+            ensureOpen();
+            duplicatedByteBuf = byteBuf.duplicate();
+            resourceHandle = () -> {
+                if (duplicatedByteBuf != null) {
+                    duplicatedByteBuf.release();
+                    duplicatedByteBuf = null;
+                }
+            };
+            trackedResources.add(resourceHandle);
             bsonReader = new BsonBinaryReader(new ByteBufferBsonInput(duplicatedByteBuf));
             // While one might expect that this would be a call to BsonReader#readStartArray that doesn't work because BsonBinaryReader
             // expects to be positioned at the start at the beginning of a document, not an array.  Fortunately, a BSON array has exactly
@@ -283,7 +342,11 @@ private class ByteBufBsonArrayIterator implements Iterator<BsonValue> {
 
         @Override
         public boolean hasNext() {
-            return bsonReader.getCurrentBsonType() != BsonType.END_OF_DOCUMENT;
+            boolean hasNext = bsonReader.getCurrentBsonType() != BsonType.END_OF_DOCUMENT;
+            if (!hasNext) {
+                cleanup();
+            }
+            return hasNext;
         }
 
         @Override
@@ -292,9 +355,22 @@ public BsonValue next() {
                 throw new NoSuchElementException();
             }
             bsonReader.skipName();
-            BsonValue value = readBsonValue(duplicatedByteBuf, bsonReader);
+            BsonValue value = readBsonValue(duplicatedByteBuf, bsonReader, trackedResources);
             bsonReader.readBsonType();
             return value;
         }
+
+        private void cleanup() {
+            if (!finished) {
+                finished = true;
+                // Remove from tracked resources since we're cleaning up immediately
+                trackedResources.remove(resourceHandle);
+                try {
+                    resourceHandle.close();
+                } catch (Exception e) {
+                    // Ignore
+                }
+            }
+        }
     }
 }