perf: Add support for virtual threads

google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/Adapter.java

@@ -15,6 +15,8 @@
 */
 package com.google.cloud.spanner.adapter;
 
+import static com.google.cloud.spanner.adapter.util.ThreadFactoryUtil.tryCreateVirtualThreadPerTaskExecutor;
+
 import com.google.api.gax.core.CredentialsProvider;
 import com.google.api.gax.core.FixedCredentialsProvider;
 import com.google.api.gax.core.GaxProperties;
@@ -95,6 +97,11 @@ void start() {
       InstantiatingGrpcChannelProvider.Builder channelProviderBuilder =
           AdapterSettings.defaultGrpcTransportProviderBuilder();
 
+      if (options.getUseVirtualThreads()) {
+        executor = tryCreateVirtualThreadPerTaskExecutor("spanner-virtual-thread");
+        channelProviderBuilder.setExecutor(executor);
+      }
+
       channelProviderBuilder
           .setAllowNonDefaultServiceAccount(true)
           .setChannelPoolSettings(
@@ -141,7 +148,9 @@ void start() {
               options.getTcpPort(), DEFAULT_CONNECTION_BACKLOG, options.getInetAddress());
       LOG.info("Local TCP server started on {}:{}", options.getInetAddress(), options.getTcpPort());
 
-      executor = Executors.newCachedThreadPool();
+      if (executor == null) {
+        executor = Executors.newCachedThreadPool();
+      }
 
       // Start accepting client connections.
       executor.execute(this::acceptClientConnections);

google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/AdapterOptions.java

@@ -37,6 +37,7 @@ static class Builder {
     Optional<Duration> maxCommitDelay = Optional.empty();
     private TransportChannelProvider channelProvider = null;
     private Credentials credentials;
+    private boolean useVirtualThreads = false;
 
     /** The Cloud Spanner endpoint. */
     Builder spannerEndpoint(String spannerEndpoint) {
@@ -92,6 +93,12 @@ Builder credentials(Credentials credentials) {
       return this;
     }
 
+    /** (Optional) Whether to use virtual threads (Java 21+ only) */
+    Builder useVirtualThreads(boolean useVirtualThreads) {
+      this.useVirtualThreads = useVirtualThreads;
+      return this;
+    }
+
     AdapterOptions build() {
       return new AdapterOptions(this);
     }
@@ -105,6 +112,7 @@ AdapterOptions build() {
   private final Optional<Duration> maxCommitDelay;
   private TransportChannelProvider channelProvider;
   private Credentials credentials;
+  private boolean useVirtualThreads;
 
   private AdapterOptions(Builder builder) {
     this.spannerEndpoint = builder.spannerEndpoint;
@@ -115,6 +123,7 @@ private AdapterOptions(Builder builder) {
     this.maxCommitDelay = builder.maxCommitDelay;
     this.channelProvider = builder.channelProvider;
     this.credentials = builder.credentials;
+    this.useVirtualThreads = builder.useVirtualThreads;
   }
 
   static Builder newBuilder() {
@@ -149,6 +158,10 @@ Credentials getCredentials() {
     return credentials;
   }
 
+  boolean getUseVirtualThreads() {
+    return useVirtualThreads;
+  }
+
   Optional<Duration> getMaxCommitDelay() {
     return maxCommitDelay;
   }

google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/Launcher.java

@@ -102,5 +102,11 @@ public static void main(String[] args) throws Exception {
         maxCommitDelayProperty);
 
     adapter.start();
+
+    try {
+      Thread.currentThread().join();
+    } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
+    }
   }
 }

google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/SpannerCqlSessionBuilder.java

@@ -58,6 +58,7 @@ public final class SpannerCqlSessionBuilder
   private Optional<Duration> maxCommitDelay = Optional.empty();
   private TransportChannelProvider channelProvider = null;
   private Credentials credentials;
+  private boolean useVirtualThreads;
 
   /**
    * Wraps the default CQL session with a SpannerCqlSession instance.
@@ -120,6 +121,16 @@ public SpannerCqlSessionBuilder setGoogleCloudCredentials(Credentials credential
     return this;
   }
 
+  /**
+   * (Optional, default `false`) Enables/disables the use of virtual threads for the gRPC executor.
+   * Setting this option only has any effect on Java 21 and higher. In all other cases, the option
+   * will be ignored.
+   */
+  protected SpannerCqlSessionBuilder setUseVirtualThreads(boolean useVirtualThreads) {
+    this.useVirtualThreads = useVirtualThreads;
+    return this;
+  }
+
   /**
    * Creates the session with the options set by this builder.
    *
@@ -231,6 +242,7 @@ private void createAndStartAdapter() {
             .numGrpcChannels(numGrpcChannels)
             .channelProvider(channelProvider)
             .credentials(credentials)
+            .useVirtualThreads(useVirtualThreads)
             .build();
 
     adapter = new Adapter(adapterOptions);

google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/util/ThreadFactoryUtil.java

@@ -0,0 +1,86 @@
+/*
+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.spanner.adapter.util;
+
+import com.google.api.core.InternalApi;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
+import javax.annotation.Nullable;
+
+/** Utility class for creating a thread factory for daemon or virtual threads. */
+@InternalApi
+public final class ThreadFactoryUtil {
+
+  /**
+   * Tries to create a {@link ThreadFactory} that creates virtual threads. Returns null if virtual
+   * threads are not supported on this JVM.
+   */
+  @InternalApi
+  @Nullable
+  public static ThreadFactory tryCreateVirtualThreadFactory(String baseNameFormat) {
+    try {
+      Class<?> threadBuilderClass = Class.forName("java.lang.Thread$Builder");
+      Method ofVirtualMethod = Thread.class.getDeclaredMethod("ofVirtual");
+      Object virtualBuilder = ofVirtualMethod.invoke(null);
+      Method nameMethod = threadBuilderClass.getDeclaredMethod("name", String.class, long.class);
+      virtualBuilder = nameMethod.invoke(virtualBuilder, baseNameFormat + "-", 0);
+      Method factoryMethod = threadBuilderClass.getDeclaredMethod("factory");
+      return (ThreadFactory) factoryMethod.invoke(virtualBuilder);
+    } catch (ClassNotFoundException | NoSuchMethodException ignore) {
+      return null;
+    } catch (InvocationTargetException | IllegalAccessException e) {
+      // Java 20 supports virtual threads as an experimental feature. It will throw an
+      // UnsupportedOperationException if experimental features have not been enabled.
+      if (e.getCause() instanceof UnsupportedOperationException) {
+        return null;
+      }
+      throw new RuntimeException(e);
+    }
+  }
+
+  /**
+   * Tries to create an {@link ExecutorService} that creates a new virtual thread for each task that
+   * it runs. Creating a new virtual thread is the recommended way to create executors using virtual
+   * threads, instead of creating a pool of virtual threads. Returns null if virtual threads are not
+   * supported on this JVM.
+   */
+  @InternalApi
+  @Nullable
+  public static ExecutorService tryCreateVirtualThreadPerTaskExecutor(String baseNameFormat) {
+    ThreadFactory factory = tryCreateVirtualThreadFactory(baseNameFormat);
+    if (factory != null) {
+      try {
+        Method newThreadPerTaskExecutorMethod =
+            Executors.class.getDeclaredMethod("newThreadPerTaskExecutor", ThreadFactory.class);
+        return (ExecutorService) newThreadPerTaskExecutorMethod.invoke(null, factory);
+      } catch (NoSuchMethodException ignore) {
+        return null;
+      } catch (InvocationTargetException | IllegalAccessException e) {
+        // Java 20 supports virtual threads as an experimental feature. It will throw an
+        // UnsupportedOperationException if experimental features have not been enabled.
+        if (e.getCause() instanceof UnsupportedOperationException) {
+          return null;
+        }
+        throw new RuntimeException(e);
+      }
+    }
+    return null;
+  }
+}

google-cloud-spanner-cassandra/src/test/java/com/google/cloud/spanner/adapter/ThreadFactoryUtilTest.java

@@ -0,0 +1,57 @@
+/*
+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.spanner.adapter;
+
+import static com.google.cloud.spanner.adapter.util.ThreadFactoryUtil.tryCreateVirtualThreadFactory;
+import static com.google.cloud.spanner.adapter.util.ThreadFactoryUtil.tryCreateVirtualThreadPerTaskExecutor;
+import static com.google.common.truth.Truth.assertThat;
+
+import java.util.concurrent.ThreadFactory;
+import org.junit.Test;
+
+public class ThreadFactoryUtilTest {
+
+  @Test
+  public void testCreateThreadFactory() throws Exception {
+    if (isJava21OrHigher()) {
+      ThreadFactory virtualFactory = tryCreateVirtualThreadFactory("test-thread");
+      assertThat(virtualFactory).isNotNull();
+    } else {
+      assertThat(tryCreateVirtualThreadFactory("test-thread")).isNull();
+    }
+  }
+
+  @Test
+  public void testTryCreateVirtualThreadPerTaskExecutor() {
+    if (isJava21OrHigher()) {
+      assertThat(tryCreateVirtualThreadPerTaskExecutor("test-virtual-thread")).isNotNull();
+    } else {
+      assertThat(tryCreateVirtualThreadPerTaskExecutor("test-virtual-thread")).isNull();
+    }
+  }
+
+  private static boolean isJava21OrHigher() {
+    String[] versionElements = System.getProperty("java.version").split("\\.");
+    int majorVersion = Integer.parseInt(versionElements[0]);
+    // Java 1.8 (Java 8) and lower used the format 1.8 etc.
+    // Java 9 and higher use the format 9.x
+    if (majorVersion == 1) {
+      majorVersion = Integer.parseInt(versionElements[1]);
+    }
+    return majorVersion >= 21;
+  }
+}