perf: withstand connect storms

src/main/java/com/google/cloud/spanner/pgadapter/ConnectionHandler.java

@@ -438,7 +438,11 @@ && getServer().getOptions().getSslMode().isSslEnabled()
           () ->
               String.format(
                   "Exception on connection handler with ID %s for client %s: %s",
-                  getName(), socket.getInetAddress().getHostAddress(), e));
+                  getName(),
+                  socket == null || socket.getInetAddress() == null
+                      ? "(none)"
+                      : socket.getInetAddress().getHostAddress(),
+                  e));
     } finally {
       if (result != RunConnectionState.RESTART_WITH_SSL) {
         logger.log(

src/main/java/com/google/cloud/spanner/pgadapter/ProxyServer.java

@@ -46,7 +46,9 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
@@ -89,6 +91,13 @@ public class ProxyServer extends AbstractApiService {
   private final ConcurrentLinkedQueue<WireMessage> debugMessages = new ConcurrentLinkedQueue<>();
   private final AtomicInteger debugMessageCount = new AtomicInteger();
 
+  private final ExecutorService createConnectionHandlerExecutor =
+      new ThreadPoolExecutor(
+          /* corePoolSize = */ 1,
+          Runtime.getRuntime().availableProcessors(),
+          /* keepAliveTime = */ 10L,
+          TimeUnit.SECONDS,
+          new LinkedBlockingQueue<>());
   private final ThreadFactory threadFactory;
 
   /**
@@ -286,6 +295,7 @@ protected void doStop() {
       }
     } catch (Throwable ignore) {
     }
+    createConnectionHandlerExecutor.shutdown();
     if (openTelemetry instanceof Closeable) {
       try {
         ((Closeable) openTelemetry).close();
@@ -403,7 +413,24 @@ void runServer(
     awaitRunning();
     try {
       while (isRunning()) {
-        createConnectionHandler(serverSocket.accept());
+        Socket socket = serverSocket.accept();
+        // Hand off the creation of the connection handler to a worker thread to ensure that we
+        // continue to listen for new incoming connection requests as quickly as possible.
+        // This prevents connection timeouts if there is a large 'connect storm' (i.e. a client
+        // sends a large number of connection requests at the same time).
+        createConnectionHandlerExecutor.submit(
+            () -> {
+              try {
+                createConnectionHandler(socket);
+              } catch (SocketException socketException) {
+                logger.log(
+                    Level.WARNING,
+                    () ->
+                        String.format(
+                            "Failed to create connection on socket %s: %s.",
+                            socket, socketException));
+              }
+            });
       }
     } catch (SocketException e) {
       // This is a normal and expected exception when the server is shutting down.

src/test/java/com/google/cloud/spanner/pgadapter/JdbcSimpleModeMockServerTest.java

@@ -26,6 +26,10 @@
 import com.google.cloud.spanner.pgadapter.error.SQLState;
 import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
 import com.google.common.collect.ImmutableList;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
 import com.google.protobuf.ListValue;
 import com.google.protobuf.Value;
 import com.google.spanner.admin.database.v1.UpdateDatabaseDdlRequest;
@@ -47,8 +51,11 @@
 import java.time.LocalDateTime;
 import java.time.OffsetDateTime;
 import java.time.ZoneOffset;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.TimeZone;
+import java.util.concurrent.Callable;
+import java.util.concurrent.Executors;
 import java.util.stream.Collectors;
 import org.junit.BeforeClass;
 import org.junit.Test;
@@ -875,6 +882,29 @@ public void testImplicitBatchOfClientSideStatements() throws SQLException {
     }
   }
 
+  @Test
+  public void testConnectStorm() throws Exception {
+    int numThreads = Runtime.getRuntime().availableProcessors() * 10;
+    ListeningExecutorService service =
+        MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(numThreads));
+    List<ListenableFuture<Void>> futures = new ArrayList<>(numThreads);
+    for (int n = 0; n < numThreads; n++) {
+      futures.add(service.submit(new ConnectCallable()));
+    }
+    assertEquals(numThreads, Futures.allAsList(futures).get().size());
+  }
+
+  class ConnectCallable implements Callable<Void> {
+
+    @Override
+    public Void call() throws Exception {
+      try (Connection ignore = DriverManager.getConnection(createUrl())) {
+        // Just connect and disconnect.
+        return null;
+      }
+    }
+  }
+
   @Test
   public void testPrepareInDmlBatch() throws SQLException {
     String sql = "insert into test (id, value) values ($1, $2)";