Repository: GoogleCloudPlatform/cloud-sql-jdbc-socket-factory (java)

Issue:
Memory leak with sockets (SSLSocketImpl) in MonitoredCache
### Bug Description

After upgrading `cloud-sql-connector-jdbc-sqlserver` from `1.23.1` to version `1.24.0` we have experienced several occasional memory leaks in a couple of our applications after running for a hours or even days.

`InternalConnectorRegistry` holds a map of `unnamedConnectors`. One of the entries in that map is an instance of `MonitoredCache` that seems to have been introduced in #2056 

`MonitoredCache` has a field `List<Socket> sockets`, and when we observe the memory leak, this list has grown to a large size. In our last case, this list held 91568 instances of `sun.security.ssl.SSLSocketImpl` with a total memory consumption 3.7GB according to the heap dump. 

https://github.com/GoogleCloudPlatform/cloud-sql-jdbc-socket-factory/blob/v1.24.0/core/src/main/java/com/google/cloud/sql/core/MonitoredCache.java#L41

Applications can run for days before this happens, We have not figured out what the trigger is. (Our applications have used the `cloud-sql-connector-jdbc-sqlserver` dependency for years without issues, and the memory leak did only occur after the upgrade to `1.24.0`)

The screenshot below shows the relevant part from the heap dump taken from the application during the memory leak. It has been imported into Intellij for investigation, and it clearly shows the memory consumption in relation to `MonitoredCache.sockets`:
<img width="1001" alt="Image" src="https://github.com/user-attachments/assets/02ea7ad8-cfe3-45f2-95de-b4c498dc8830" />

Relevant parts of dependency tree
```java
[INFO] +- com.google.cloud.sql:cloud-sql-connector-jdbc-sqlserver:jar:1.24.0:compile
[INFO] |  \- com.google.cloud.sql:jdbc-socket-factory-core:jar:1.24.0:compile
[INFO] |     +- com.github.jnr:jnr-enxio:jar:0.32.18:compile
[INFO] |     \- com.google.apis:google-api-services-sqladmin:jar:v1beta4-rev20250310-2.0.0:compile
[INFO] +- com.microsoft.sqlserver:mssql-jdbc:jar:12.10.0.jre11:compile
```

Application runs in kubernetes on GKE with the base image `eclipse-temurin:21-jre-alpine`. Java application is built with jib maven plugin `3.4.4` and java compile/release version is 21. Application consumes messages from pubsub pull and writes to a mssql database using the socket factory

### Example code (or command)

```Java
// JDBC url
jdbc:sqlserver://localhost;databaseName=<databaseName>;socketFactoryConstructorArg=<dbInstance>?ipTypes=PRIVATE;socketFactoryClass=com.google.cloud.sql.sqlserver.SocketFactory;trustServerCertificate=true;encrypt=true;socketTimeout=120000;queryTimeout=30;cancelQueryTimeout=20;

// Setting up data source
  public static HikariDataSource getDataSource(final ParameterResolver parameterResolver, final String databasePassword) {
    final String dbUrl = "jdbc:sqlserver://localhost;databaseName=<databaseName>;socketFactoryConstructorArg=<dbInstance>?ipTypes=PRIVATE;socketFactoryClass=com.google.cloud.sql.sqlserver.SocketFactory;trustServerCertificate=true;encrypt=true;socketTimeout=120000;queryTimeout=30;cancelQueryTimeout=20;";
    final String dbUsername = "username";
    final HikariDataSource dataSource = new HikariDataSource();
    dataSource.setDriverClassName("com.microsoft.sqlserver.jdbc.SQLServerDriver");
    dataSource.setJdbcUrl(dbUrl);
    dataSource.setUsername(dbUsername);
    dataSource.setPassword(databasePassword);
    dataSource.setLeakDetectionThreshold(40000);
    dataSource.setKeepaliveTime(30000);
    dataSource.setMaxLifetime(60000);
    dataSource.setTransactionIsolation("TRANSACTION_READ_COMMITTED");
    final int maximumPoolSize = 20;
    dataSource.setMaximumPoolSize(maximumPoolSize);
    return dataSource;
  }
```

### Stacktrace

```bash

```

### Steps to reproduce?

It is hard to reproduce this memory leak issue. Will provide more information if possible.


### Environment

1. OS type and version: eclipse-temurin:21-jre-alpine
2. Java SDK version: OpenJDK 64-Bit Server VM Temurin-21.0.6+7 (build 21.0.6+7-LTS, mixed mode, sharing)
3. Cloud SQL Java Socket Factory version: 1.24.0


### Additional Details

_No response_

Pull request:
fix: only keep track of sockets opened using domain name
The list `sockets` on the `MonitoredCache` should only keep track of
sockets opened for connections using a domain name.

It should not keep track of all sockets for regular connections.

This is what it is currently doing:

https://github.com/GoogleCloudPlatform/cloud-sql-jdbc-socket-factory/blob/a5329ac62ce877783c447f4192b62e72f39b63d4/core/src/main/java/com/google/cloud/sql/core/Connector.java#L141

Sockets are only removed from the list if the connections were made using a domain name:

https://github.com/GoogleCloudPlatform/cloud-sql-jdbc-socket-factory/blob/a5329ac62ce877783c447f4192b62e72f39b63d4/core/src/main/java/com/google/cloud/sql/core/MonitoredCache.java#L92-L96

Thus for sockets not using a domain name and using the default instance connection name pattern, the sockets list will continue to grow and grow without being cleaned up.

This PR adds the proper check to gate adding to the `sockets` list for only connections using a domain name.

Fixes #2129 

Patch:
diff --git a/core/src/main/java/com/google/cloud/sql/core/MonitoredCache.java b/core/src/main/java/com/google/cloud/sql/core/MonitoredCache.java
index 069422ab..990c709c 100644
--- a/core/src/main/java/com/google/cloud/sql/core/MonitoredCache.java
+++ b/core/src/main/java/com/google/cloud/sql/core/MonitoredCache.java
@@ -16,15 +16,16 @@
 
 package com.google.cloud.sql.core;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Strings;
 import java.io.IOException;
 import java.net.Socket;
-import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Iterator;
-import java.util.List;
+import java.util.Set;
 import java.util.Timer;
 import java.util.TimerTask;
+import java.util.WeakHashMap;
 import java.util.function.Function;
 import javax.net.ssl.SSLSocket;
 import org.slf4j.Logger;
@@ -38,7 +39,11 @@ import org.slf4j.LoggerFactory;
 class MonitoredCache implements ConnectionInfoCache {
   private static final Logger logger = LoggerFactory.getLogger(Connector.class);
   private final ConnectionInfoCache cache;
-  private final List<Socket> sockets = Collections.synchronizedList(new ArrayList<>());
+  // Use weak references to hold the open sockets. If a socket is no longer in
+  // use by the application, the garabage collector will automatically remove
+  // it from this set.
+  private final Set<Socket> sockets =
+      Collections.synchronizedSet(Collections.newSetFromMap(new WeakHashMap<>()));
   private final Function<ConnectionConfig, CloudSqlInstanceName> resolve;
   private final TimerTask task;
 
@@ -49,6 +54,8 @@ class MonitoredCache implements ConnectionInfoCache {
     this.cache = cache;
     this.resolve = resolve;
 
+    // If this was configured with a domain name, start the domain name check
+    // and socket cleanup periodic task.
     if (!Strings.isNullOrEmpty(cache.getConfig().getDomainName())) {
       long failoverPeriod = cache.getConfig().getConnectorConfig().getFailoverPeriod().toMillis();
       this.task =
@@ -64,6 +71,11 @@ class MonitoredCache implements ConnectionInfoCache {
     }
   }
 
+  @VisibleForTesting
+  int getOpenSocketCount() {
+    return sockets.size();
+  }
+
   private void checkDomainName() {
     // Resolve the domain name again. If it changed, close the sockets
     try {
@@ -149,6 +161,10 @@ class MonitoredCache implements ConnectionInfoCache {
   }
 
   synchronized void addSocket(SSLSocket socket) {
-    sockets.add(socket);
+    // Only add the socket if this was configured using a domain name,
+    // and therefore the background socket cleanup task is running.
+    if (!Strings.isNullOrEmpty(cache.getConfig().getDomainName())) {
+      sockets.add(socket);
+    }
   }
 }


Test patch:
diff --git a/core/src/test/java/com/google/cloud/sql/core/MonitoredCacheTest.java b/core/src/test/java/com/google/cloud/sql/core/MonitoredCacheTest.java
new file mode 100644
index 00000000..47d78eb1
--- /dev/null
+++ b/core/src/test/java/com/google/cloud/sql/core/MonitoredCacheTest.java
@@ -0,0 +1,208 @@
+/*
+ * Copyright 2025 Google LLC
+ *
+ * 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.google.cloud.sql.core;
+
+import com.google.cloud.sql.ConnectorConfig;
+import java.io.IOException;
+import java.time.Duration;
+import java.util.Timer;
+import javax.net.ssl.HandshakeCompletedListener;
+import javax.net.ssl.SSLSession;
+import javax.net.ssl.SSLSocket;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class MonitoredCacheTest {
+  private static final Timer timer = new Timer(true);
+
+  @AfterClass
+  public static void afterClass() {
+    timer.cancel();
+  }
+
+  @Test
+  public void testMonitoredCacheHoldsSocketsWithDomainName() {
+    CloudSqlInstanceName name = new CloudSqlInstanceName("proj:reg:inst", "db.example.com");
+    ConnectionConfig config =
+        new ConnectionConfig.Builder()
+            .withCloudSqlInstance("proj:reg:inst")
+            .withDomainName("db.example.com")
+            .build();
+    MockCache mockCache = new MockCache(config);
+
+    MonitoredCache cache = new MonitoredCache(mockCache, timer, connectionConfig -> name);
+    MockSslSocket socket = new MockSslSocket();
+    cache.addSocket(socket);
+    Assert.assertEquals("1 socket in cache", 1, cache.getOpenSocketCount());
+    cache.close();
+    Assert.assertTrue("socket closed", socket.closed);
+  }
+
+  @Test
+  public void testMonitoredCachePurgesClosedSockets() throws InterruptedException {
+    CloudSqlInstanceName name = new CloudSqlInstanceName("proj:reg:inst", "db.example.com");
+    // Purge sockets every 10ms.
+    ConnectionConfig config =
+        new ConnectionConfig.Builder()
+            .withCloudSqlInstance("proj:reg:inst")
+            .withDomainName("db.example.com")
+            .withConnectorConfig(
+                new ConnectorConfig.Builder().withFailoverPeriod(Duration.ofMillis(10)).build())
+            .build();
+    MockCache mockCache = new MockCache(config);
+
+    MonitoredCache cache = new MonitoredCache(mockCache, timer, connectionConfig -> name);
+    MockSslSocket socket = new MockSslSocket();
+    cache.addSocket(socket);
+    Assert.assertEquals("1 socket in cache", 1, cache.getOpenSocketCount());
+    socket.close();
+    Thread.sleep(20);
+    Assert.assertEquals("0 socket in cache", 0, cache.getOpenSocketCount());
+  }
+
+  @Test
+  public void testMonitoredCacheWithoutDomainNameIgnoresSockets() {
+    CloudSqlInstanceName name = new CloudSqlInstanceName("proj:reg:inst");
+    ConnectionConfig config =
+        new ConnectionConfig.Builder().withCloudSqlInstance("proj:reg:inst").build();
+    MockCache mockCache = new MockCache(config);
+
+    MonitoredCache cache = new MonitoredCache(mockCache, timer, connectionConfig -> name);
+    MockSslSocket socket = new MockSslSocket();
+    cache.addSocket(socket);
+    Assert.assertEquals("0 socket in cache", 0, cache.getOpenSocketCount());
+  }
+
+  private static class MockSslSocket extends SSLSocket {
+    boolean closed;
+
+    @Override
+    public synchronized boolean isClosed() {
+      return closed;
+    }
+
+    @Override
+    public synchronized void close() {
+      this.closed = true;
+    }
+
+    @Override
+    public String[] getSupportedCipherSuites() {
+      return new String[0];
+    }
+
+    @Override
+    public String[] getEnabledCipherSuites() {
+      return new String[0];
+    }
+
+    @Override
+    public void setEnabledCipherSuites(String[] suites) {}
+
+    @Override
+    public String[] getSupportedProtocols() {
+      return new String[0];
+    }
+
+    @Override
+    public String[] getEnabledProtocols() {
+      return new String[0];
+    }
+
+    @Override
+    public void setEnabledProtocols(String[] protocols) {}
+
+    @Override
+    public SSLSession getSession() {
+      return null;
+    }
+
+    @Override
+    public void addHandshakeCompletedListener(HandshakeCompletedListener listener) {}
+
+    @Override
+    public void removeHandshakeCompletedListener(HandshakeCompletedListener listener) {}
+
+    @Override
+    public void startHandshake() throws IOException {}
+
+    @Override
+    public void setUseClientMode(boolean mode) {}
+
+    @Override
+    public boolean getUseClientMode() {
+      return false;
+    }
+
+    @Override
+    public void setNeedClientAuth(boolean need) {}
+
+    @Override
+    public boolean getNeedClientAuth() {
+      return false;
+    }
+
+    @Override
+    public void setWantClientAuth(boolean want) {}
+
+    @Override
+    public boolean getWantClientAuth() {
+      return false;
+    }
+
+    @Override
+    public void setEnableSessionCreation(boolean flag) {}
+
+    @Override
+    public boolean getEnableSessionCreation() {
+      return false;
+    }
+  }
+
+  private static class MockCache implements ConnectionInfoCache {
+    private final ConnectionConfig config;
+
+    MockCache(ConnectionConfig config) {
+      this.config = config;
+    }
+
+    @Override
+    public ConnectionMetadata getConnectionMetadata(long timeoutMs) {
+      return null;
+    }
+
+    @Override
+    public void forceRefresh() {}
+
+    @Override
+    public void refreshIfExpired() {}
+
+    @Override
+    public void close() {}
+
+    @Override
+    public boolean isClosed() {
+      return false;
+    }
+
+    @Override
+    public ConnectionConfig getConfig() {
+      return config;
+    }
+  }
+}