diff --git a/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/Adapter.java b/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/Adapter.java index 6b9ccc37..aaa06bb8 100644 --- a/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/Adapter.java +++ b/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); diff --git a/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/AdapterOptions.java b/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/AdapterOptions.java index 65857b11..0bd2431b 100644 --- a/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/AdapterOptions.java +++ b/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/AdapterOptions.java @@ -37,6 +37,7 @@ static class Builder { Optional 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 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 getMaxCommitDelay() { return maxCommitDelay; } diff --git a/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/Launcher.java b/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/Launcher.java index b7c0efb4..edcf299e 100644 --- a/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/Launcher.java +++ b/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(); + } } } diff --git a/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/SpannerCqlSessionBuilder.java b/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/SpannerCqlSessionBuilder.java index e4e9a4f7..06bf26c8 100644 --- a/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/SpannerCqlSessionBuilder.java +++ b/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/SpannerCqlSessionBuilder.java @@ -58,6 +58,7 @@ public final class SpannerCqlSessionBuilder private Optional 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); diff --git a/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/util/ThreadFactoryUtil.java b/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/util/ThreadFactoryUtil.java new file mode 100644 index 00000000..b4bfed0b --- /dev/null +++ b/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; + } +} diff --git a/google-cloud-spanner-cassandra/src/test/java/com/google/cloud/spanner/adapter/ThreadFactoryUtilTest.java b/google-cloud-spanner-cassandra/src/test/java/com/google/cloud/spanner/adapter/ThreadFactoryUtilTest.java new file mode 100644 index 00000000..d891747d --- /dev/null +++ b/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; + } +}