From b8dfe8c450a0c148a20576e2d84eace939c26a55 Mon Sep 17 00:00:00 2001 From: Ludovic Champenois Date: Wed, 2 Sep 2026 15:11:57 -0700 Subject: [PATCH 1/3] Optimize API clients and Jetty runtime for Virtual Threads and high throughput This change modernizes thread handling across API clients and the Jetty runtime, adding support and protections for Java 21+ Virtual Threads: 1. Bounded Concurrency for Virtual Threads in JdkHttpApiHostClient: - When appengine.api.use.virtualthreads is enabled, requests execute on virtual threads, but in-flight HTTP calls to the API host are capped using a Semaphore bounded by maxThreads (APPENGINE_API_MAX_THREADS / APPENGINE_API_MAX_CONNECTIONS, default 100). - This prevents unconstrained concurrency bursts and retry storms from saturating the Appserver/Datastore proxy while benefiting from virtual threads. - Preserves JDK 17 compilation and execution compatibility via reflective lookup of newVirtualThreadPerTaskExecutor(). 2. Bounded Thread Pools in JettyHttpApiHostClient & JdkHttpApiHostClient: - Caps thread pools at maxThreads (default 100) instead of unbounded/200, safely queueing requests during transient backend degradation to avoid triggering masked INTERNAL_ERROR ("Internal Datastore Error") responses. 3. Dynamic JVM Runtime Capability Checks: - Updates JettyServletEngineAdapter (Jetty 12 and 12.1) and JettyContainerService (local devappserver) to dynamically check Runtime.version().feature() >= 21 when activating virtual thread pools, ensuring forward compatibility with Java 25+ without hardcoded runtime string dependencies. 4. VirtualThreadSupport Utility Classes: - Introduces VirtualThreadSupport in api, api_dev, and runtime/impl using MethodHandles to dynamically detect virtual thread capability and create unstarted virtual threads on JDK 21+ while remaining fully compatible with Java 17 compile targets. 5. Datastore Transaction Backoff Retries: - Implements exponential backoff retries in DatastoreServiceImpl.beginTransaction() controlled by appengine.datastore.retries (default 1). 6. Documentation: - Adds runtime/runtime_impl_jetty121/API_CLIENTS.md detailing all configuration properties, thread pool bounds, virtual thread behavior, and Datastore retry settings. --- .../api/datastore/DatastoreServiceImpl.java | 28 +++- .../appengine/setup/VirtualThreadSupport.java | 93 +++++++++++ .../development/VirtualThreadSupport.java | 93 +++++++++++ .../utils/runtime/VirtualThreadSupport.java | 93 +++++++++++ .../jetty/JettyContainerService.java | 4 +- .../jetty/ee11/JettyContainerService.java | 4 +- .../jetty/JettyServletEngineAdapter.java | 6 +- runtime/runtime_impl_jetty121/API_CLIENTS.md | 151 ++++++++++++++++++ .../runtime/http/HttpApiHostClient.java | 24 +++ .../http/HttpApiHostClientFactory.java | 23 +++ .../runtime/http/JdkHttpApiHostClient.java | 113 +++++++++++-- .../runtime/http/JettyHttpApiHostClient.java | 75 ++++++--- .../jetty/JettyServletEngineAdapter.java | 6 +- 13 files changed, 674 insertions(+), 39 deletions(-) create mode 100644 api/src/main/java/com/google/appengine/setup/VirtualThreadSupport.java create mode 100644 api_dev/src/main/java/com/google/appengine/tools/development/VirtualThreadSupport.java create mode 100644 runtime/impl/src/main/java/com/google/apphosting/utils/runtime/VirtualThreadSupport.java create mode 100644 runtime/runtime_impl_jetty121/API_CLIENTS.md diff --git a/api/src/main/java/com/google/appengine/api/datastore/DatastoreServiceImpl.java b/api/src/main/java/com/google/appengine/api/datastore/DatastoreServiceImpl.java index d0444ae6b..1e0d0d413 100644 --- a/api/src/main/java/com/google/appengine/api/datastore/DatastoreServiceImpl.java +++ b/api/src/main/java/com/google/appengine/api/datastore/DatastoreServiceImpl.java @@ -18,6 +18,7 @@ import static com.google.appengine.api.datastore.FutureHelper.quietGet; +import com.google.apphosting.api.ApiProxy; import java.util.Collection; import java.util.List; import java.util.Map; @@ -30,6 +31,8 @@ final class DatastoreServiceImpl implements DatastoreService { private final AsyncDatastoreServiceInternal async; + static final long BEGIN_TXN_RETRY_DELAY_MS = 100; + private static final int MAX_RETRIES = Integer.getInteger("appengine.datastore.retries", 1); public DatastoreServiceImpl(AsyncDatastoreServiceInternal async) { this.async = async; @@ -137,12 +140,33 @@ public KeyRangeState allocateIdRange(KeyRange range) { @Override public Transaction beginTransaction() { - return quietGet(async.beginTransaction()); + return beginTransaction(TransactionOptions.Builder.withDefaults()); } @Override public Transaction beginTransaction(TransactionOptions options) { - return quietGet(async.beginTransaction(options)); + int retries = 0; + long delay = BEGIN_TXN_RETRY_DELAY_MS; + while (true) { + try { + Transaction tx = quietGet(async.beginTransaction(options)); + tx.getId(); // Force handle resolution + return tx; + } catch (DatastoreFailureException + | DatastoreTimeoutException + | ApiProxy.RPCFailedException e) { + if (++retries > MAX_RETRIES) { + throw e; + } + try { + Thread.sleep(delay); + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + throw e; + } + delay *= 2; + } + } } @Override diff --git a/api/src/main/java/com/google/appengine/setup/VirtualThreadSupport.java b/api/src/main/java/com/google/appengine/setup/VirtualThreadSupport.java new file mode 100644 index 000000000..361187c27 --- /dev/null +++ b/api/src/main/java/com/google/appengine/setup/VirtualThreadSupport.java @@ -0,0 +1,93 @@ +/* + * Copyright 2022 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 + * + * https://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.appengine.setup; + +import java.lang.invoke.MethodHandle; +import java.lang.invoke.MethodHandles; +import java.lang.invoke.MethodType; + +/** + * Utility class to support virtual threads using reflection, enabling compatibility with JDK 17. + */ +public class VirtualThreadSupport { + private static final MethodHandle OF_VIRTUAL; + private static final MethodHandle UNSTARTED; + private static final MethodHandle IS_VIRTUAL; + + static { + MethodHandle ofVirtual = null; + MethodHandle unstarted = null; + MethodHandle isVirtual = null; + try { + MethodHandles.Lookup lookup = MethodHandles.publicLookup(); + Class threadClass = Thread.class; + // In JDK 21+, Thread.ofVirtual() returns a Thread.Builder.OfVirtual + Class builderClass = Class.forName("java.lang.Thread$Builder$OfVirtual"); + + ofVirtual = lookup.findStatic(threadClass, "ofVirtual", MethodType.methodType(builderClass)); + unstarted = lookup.findVirtual(builderClass, "unstarted", MethodType.methodType(threadClass, Runnable.class)); + isVirtual = lookup.findVirtual(threadClass, "isVirtual", MethodType.methodType(boolean.class)); + } catch (ClassNotFoundException | NoSuchMethodException | IllegalAccessException ignored) { + // Not on JDK 21+ or virtual threads not available + } + OF_VIRTUAL = ofVirtual; + UNSTARTED = unstarted; + IS_VIRTUAL = isVirtual; + } + + /** + * Returns true if virtual threads are supported by the current JVM. + */ + public static boolean isSupported() { + return OF_VIRTUAL != null; + } + + /** + * Creates an unstarted virtual thread if supported. + * + * @param runnable the runnable to execute + * @return an unstarted virtual thread, or null if not supported + */ + public static Thread createVirtualThread(Runnable runnable) { + if (OF_VIRTUAL == null || UNSTARTED == null) { + return null; + } + try { + Object builder = OF_VIRTUAL.invoke(); + return (Thread) UNSTARTED.invoke(builder, runnable); + } catch (Throwable t) { + return null; + } + } + + /** + * Checks if the given thread is a virtual thread. + * + * @param thread the thread to check + * @return true if the thread is virtual, false otherwise + */ + public static boolean isVirtual(Thread thread) { + if (IS_VIRTUAL == null) { + return false; + } + try { + return (boolean) IS_VIRTUAL.invoke(thread); + } catch (Throwable t) { + return false; + } + } +} diff --git a/api_dev/src/main/java/com/google/appengine/tools/development/VirtualThreadSupport.java b/api_dev/src/main/java/com/google/appengine/tools/development/VirtualThreadSupport.java new file mode 100644 index 000000000..79f65f35a --- /dev/null +++ b/api_dev/src/main/java/com/google/appengine/tools/development/VirtualThreadSupport.java @@ -0,0 +1,93 @@ +/* + * Copyright 2021 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 + * + * https://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.appengine.tools.development; + +import java.lang.invoke.MethodHandle; +import java.lang.invoke.MethodHandles; +import java.lang.invoke.MethodType; + +/** + * Utility class to support virtual threads using reflection, enabling compatibility with JDK 17. + */ +public class VirtualThreadSupport { + private static final MethodHandle OF_VIRTUAL; + private static final MethodHandle UNSTARTED; + private static final MethodHandle IS_VIRTUAL; + + static { + MethodHandle ofVirtual = null; + MethodHandle unstarted = null; + MethodHandle isVirtual = null; + try { + MethodHandles.Lookup lookup = MethodHandles.publicLookup(); + Class threadClass = Thread.class; + // In JDK 21+, Thread.ofVirtual() returns a Thread.Builder.OfVirtual + Class builderClass = Class.forName("java.lang.Thread$Builder$OfVirtual"); + + ofVirtual = lookup.findStatic(threadClass, "ofVirtual", MethodType.methodType(builderClass)); + unstarted = lookup.findVirtual(builderClass, "unstarted", MethodType.methodType(threadClass, Runnable.class)); + isVirtual = lookup.findVirtual(threadClass, "isVirtual", MethodType.methodType(boolean.class)); + } catch (ClassNotFoundException | NoSuchMethodException | IllegalAccessException ignored) { + // Not on JDK 21+ or virtual threads not available + } + OF_VIRTUAL = ofVirtual; + UNSTARTED = unstarted; + IS_VIRTUAL = isVirtual; + } + + /** + * Returns true if virtual threads are supported by the current JVM. + */ + public static boolean isSupported() { + return OF_VIRTUAL != null; + } + + /** + * Creates an unstarted virtual thread if supported. + * + * @param runnable the runnable to execute + * @return an unstarted virtual thread, or null if not supported + */ + public static Thread createVirtualThread(Runnable runnable) { + if (OF_VIRTUAL == null || UNSTARTED == null) { + return null; + } + try { + Object builder = OF_VIRTUAL.invoke(); + return (Thread) UNSTARTED.invoke(builder, runnable); + } catch (Throwable t) { + return null; + } + } + + /** + * Checks if the given thread is a virtual thread. + * + * @param thread the thread to check + * @return true if the thread is virtual, false otherwise + */ + public static boolean isVirtual(Thread thread) { + if (IS_VIRTUAL == null) { + return false; + } + try { + return (boolean) IS_VIRTUAL.invoke(thread); + } catch (Throwable t) { + return false; + } + } +} diff --git a/runtime/impl/src/main/java/com/google/apphosting/utils/runtime/VirtualThreadSupport.java b/runtime/impl/src/main/java/com/google/apphosting/utils/runtime/VirtualThreadSupport.java new file mode 100644 index 000000000..2fc126e9f --- /dev/null +++ b/runtime/impl/src/main/java/com/google/apphosting/utils/runtime/VirtualThreadSupport.java @@ -0,0 +1,93 @@ +/* + * Copyright 2021 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 + * + * https://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.apphosting.utils.runtime; + +import java.lang.invoke.MethodHandle; +import java.lang.invoke.MethodHandles; +import java.lang.invoke.MethodType; + +/** + * Utility class to support virtual threads using reflection, enabling compatibility with JDK 17. + */ +public class VirtualThreadSupport { + private static final MethodHandle OF_VIRTUAL; + private static final MethodHandle UNSTARTED; + private static final MethodHandle IS_VIRTUAL; + + static { + MethodHandle ofVirtual = null; + MethodHandle unstarted = null; + MethodHandle isVirtual = null; + try { + MethodHandles.Lookup lookup = MethodHandles.publicLookup(); + Class threadClass = Thread.class; + // In JDK 21+, Thread.ofVirtual() returns a Thread.Builder.OfVirtual + Class builderClass = Class.forName("java.lang.Thread$Builder$OfVirtual"); + + ofVirtual = lookup.findStatic(threadClass, "ofVirtual", MethodType.methodType(builderClass)); + unstarted = lookup.findVirtual(builderClass, "unstarted", MethodType.methodType(threadClass, Runnable.class)); + isVirtual = lookup.findVirtual(threadClass, "isVirtual", MethodType.methodType(boolean.class)); + } catch (ClassNotFoundException | NoSuchMethodException | IllegalAccessException ignored) { + // Not on JDK 21+ or virtual threads not available + } + OF_VIRTUAL = ofVirtual; + UNSTARTED = unstarted; + IS_VIRTUAL = isVirtual; + } + + /** + * Returns true if virtual threads are supported by the current JVM. + */ + public static boolean isSupported() { + return OF_VIRTUAL != null; + } + + /** + * Creates an unstarted virtual thread if supported. + * + * @param runnable the runnable to execute + * @return an unstarted virtual thread, or null if not supported + */ + public static Thread createVirtualThread(Runnable runnable) { + if (OF_VIRTUAL == null || UNSTARTED == null) { + return null; + } + try { + Object builder = OF_VIRTUAL.invoke(); + return (Thread) UNSTARTED.invoke(builder, runnable); + } catch (Throwable t) { + return null; + } + } + + /** + * Checks if the given thread is a virtual thread. + * + * @param thread the thread to check + * @return true if the thread is virtual, false otherwise + */ + public static boolean isVirtual(Thread thread) { + if (IS_VIRTUAL == null) { + return false; + } + try { + return (boolean) IS_VIRTUAL.invoke(thread); + } catch (Throwable t) { + return false; + } + } +} diff --git a/runtime/local_jetty121/src/main/java/com/google/appengine/tools/development/jetty/JettyContainerService.java b/runtime/local_jetty121/src/main/java/com/google/appengine/tools/development/jetty/JettyContainerService.java index ff4be0db5..a857e8148 100644 --- a/runtime/local_jetty121/src/main/java/com/google/appengine/tools/development/jetty/JettyContainerService.java +++ b/runtime/local_jetty121/src/main/java/com/google/appengine/tools/development/jetty/JettyContainerService.java @@ -339,8 +339,8 @@ protected void connectContainer() throws Exception { configuration.setSendDateHeader(false); configuration.setSendServerVersion(false); configuration.setSendXPoweredBy(false); - // Try to enable virtual threads if requested on java21: - if (Boolean.getBoolean("appengine.use.virtualthreads")) { + // Try to enable virtual threads if requested on Java 21+: + if (Boolean.getBoolean("appengine.use.virtualthreads") && Runtime.version().feature() >= 21) { QueuedThreadPool threadPool = new QueuedThreadPool(); threadPool.setVirtualThreadsExecutor(VirtualThreads.getDefaultVirtualThreadsExecutor()); server = new Server(threadPool); diff --git a/runtime/local_jetty121_ee11/src/main/java/com/google/appengine/tools/development/jetty/ee11/JettyContainerService.java b/runtime/local_jetty121_ee11/src/main/java/com/google/appengine/tools/development/jetty/ee11/JettyContainerService.java index 5f040c6e9..21650e68d 100644 --- a/runtime/local_jetty121_ee11/src/main/java/com/google/appengine/tools/development/jetty/ee11/JettyContainerService.java +++ b/runtime/local_jetty121_ee11/src/main/java/com/google/appengine/tools/development/jetty/ee11/JettyContainerService.java @@ -344,8 +344,8 @@ protected void connectContainer() throws Exception { configuration.setSendDateHeader(false); configuration.setSendServerVersion(false); configuration.setSendXPoweredBy(false); - // Try to enable virtual threads if requested on java21: - if (Boolean.getBoolean("appengine.use.virtualthreads")) { + // Try to enable virtual threads if requested on Java 21+: + if (Boolean.getBoolean("appengine.use.virtualthreads") && Runtime.version().feature() >= 21) { QueuedThreadPool threadPool = new QueuedThreadPool(); threadPool.setVirtualThreadsExecutor(VirtualThreads.getDefaultVirtualThreadsExecutor()); server = new Server(threadPool); diff --git a/runtime/runtime_impl_jetty12/src/main/java/com/google/apphosting/runtime/jetty/JettyServletEngineAdapter.java b/runtime/runtime_impl_jetty12/src/main/java/com/google/apphosting/runtime/jetty/JettyServletEngineAdapter.java index 937911434..0daec0ee2 100644 --- a/runtime/runtime_impl_jetty12/src/main/java/com/google/apphosting/runtime/jetty/JettyServletEngineAdapter.java +++ b/runtime/runtime_impl_jetty12/src/main/java/com/google/apphosting/runtime/jetty/JettyServletEngineAdapter.java @@ -103,9 +103,11 @@ public void start(String serverInfo, ServletEngineAdapter.Config runtimeOptions) boolean isHttpConnectorMode = Boolean.getBoolean(HTTP_CONNECTOR_MODE); QueuedThreadPool threadPool = new QueuedThreadPool(MAX_THREAD_POOL_THREADS, MIN_THREAD_POOL_THREADS); - // Try to enable virtual threads if requested and on java21: + // Try to enable virtual threads if requested and on Java 21+: if (Boolean.getBoolean("appengine.use.virtualthreads") - && ("java21".equals(GAE_RUNTIME) || "java25".equals(GAE_RUNTIME))) { + && (Runtime.version().feature() >= 21 + || "java21".equals(GAE_RUNTIME) + || "java25".equals(GAE_RUNTIME))) { int maxParallelism = getMaxSafeCarrierParallelism(); Executor virtualThreadsExecutor = new ForkJoinPool( diff --git a/runtime/runtime_impl_jetty121/API_CLIENTS.md b/runtime/runtime_impl_jetty121/API_CLIENTS.md new file mode 100644 index 000000000..48af0f192 --- /dev/null +++ b/runtime/runtime_impl_jetty121/API_CLIENTS.md @@ -0,0 +1,151 @@ + +# App Engine API Client Configuration + +The App Engine Java runtime communicates with Google Cloud APIs (such as +Datastore, Task Queue, and Memcache) using an HTTP-based RPC mechanism. The +runtime includes two HTTP client implementations for this purpose: a default +client based on Jetty and an alternative client using the JDK's built-in HTTP +facilities. + +This document describes both clients and how to configure them using environment +variables and Java system properties. + +## Jetty HTTP Client (Default) + +The Jetty HTTP client is the default client used by the runtime. It is based on +the [Eclipse Jetty](https://eclipse.dev/jetty/) HTTP client and is optimized for +high performance and efficient connection management. + +By default, the client is configured to allow a maximum of 100 concurrent +threads and 100 concurrent connections for API calls. These limits help +prevent memory exhaustion on smaller App Engine instance types (like F1 or F2) +and avoid overwhelming backend services during sudden traffic spikes or high +rates of failing requests with aggressive retry logic. If the thread limit is +reached, subsequent requests are queued until a thread becomes available. + +### Configuration + +You can configure the Jetty client using the following environment variables +and system properties: + +* **`APPENGINE_API_MAX_CONNECTIONS`** (Environment Variable): Sets the + maximum number of concurrent connections in the HTTP client pool. + * Default: `100` +* **`APPENGINE_API_MAX_THREADS`** (Environment Variable): Sets the + maximum number of concurrent threads for executing API calls. If unset, + this also defaults to 100. This is the most direct way to control API + call throughput and prevent backend overload. + * Default: `100` +* **`APPENGINE_API_CALLS_IDLE_TIMEOUT_MS`** (Environment Variable): Sets + the idle timeout in milliseconds for connections in the connection pool. + Connections that are idle for longer than this duration may be closed. + * Default: `58000` (58 seconds) + +## JDK HTTP Client + +The JDK HTTP client uses Java's built-in `HttpURLConnection` for API calls. It +is provided as an alternative to the Jetty client and can be useful for +troubleshooting network- or connection-related issues that might be specific to +one client implementation. In general, it may be less performant than the +default Jetty client. + +To use the JDK client instead of the Jetty client, set the +`APPENGINE_API_CALLS_USING_JDK_CLIENT` environment variable to any non-null value +(e.g., `true`). + +### Configuration + +When using the JDK client, you can configure its behavior with the following +settings: + +* **`appengine.api.use.virtualthreads`** (Java System Property): If set to `true`, + the JDK client will use Java Virtual Threads (when available on the JVM) to + handle API requests. This avoids platform thread stack size overhead while + keeping concurrent in-flight calls safely throttled via a semaphore to prevent + backend overload and retry storms. Note that this is different than + `appengine.use.virtualthreads`. + * Default: `false` + +The JDK client controls concurrency using the following environment variables (both when using virtual threads and when using the platform thread pool): + +* **`APPENGINE_API_MAX_CONNECTIONS`** (Environment Variable): Sets the + maximum number of concurrent connections in the HTTP client pool. + * Default: `100` +* **`APPENGINE_API_MAX_THREADS`** (Environment Variable): Sets the + maximum number of concurrent API calls (via thread pool size or virtual thread concurrency semaphore). + This limits the number of concurrent in-flight API calls to prevent overwhelming + the backend. + * Default: `100` + +## Datastore-Specific Configuration + +### `beginTransaction` Retries + +In addition to the HTTP client settings, the Datastore client library includes +specific retry logic for `beginTransaction()` calls. These calls can fail with +transient errors such as `DatastoreFailureException`, `DatastoreTimeoutException`, +or `ApiProxy.RPCFailedException`, especially under high contention when +multiple transactions attempt to access the same entity group simultaneously. + +To handle this, `DatastoreService.beginTransaction()` automatically retries +failed attempts with exponential backoff, starting at 100ms. You can +configure the number of retry attempts using a system property: + +* **`appengine.datastore.retries`** (Java System Property): The maximum + number of times to retry a `beginTransaction` call if it fails with + `DatastoreFailureException`, `DatastoreTimeoutException`, or + `ApiProxy.RPCFailedException`. This retry logic applies only to + `beginTransaction` calls; other Datastore operations are not retried + by this mechanism. + * Default: `1` + + +```xml + + + +``` + +## Recommended value for Jetty 12.1 / Java 25 +The Java 25 runtime environment is highly performant, and in rare cases of very high throughput, +this can lead to backend services (like Datastore) being temporarily overloaded, +which may result in exceptions like `DatastoreFailureException: Internal Datastore Error`. +If you encounter such issues, you can throttle the rate of API calls by reducing +the maximum number of concurrent threads used by the API client. +We recommend starting with a lower value and adjusting as needed: +`APPENGINE_API_MAX_THREADS=50` + +## Configuring via `appengine-web.xml` + +You can set these options by adding `` and +`` sections to your `appengine-web.xml` file. + +For example, to switch to the JDK client, enable virtual threads, increase +the thread limit to 200, and set Datastore `beginTransaction` retries to 3, +you would add: + +```xml + + + + + + + + + +``` diff --git a/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/HttpApiHostClient.java b/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/HttpApiHostClient.java index 8d44a665d..d5426e36e 100644 --- a/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/HttpApiHostClient.java +++ b/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/HttpApiHostClient.java @@ -76,6 +76,12 @@ abstract class HttpApiHostClient implements APIHostClientInterface { abstract static class Config { abstract double extraTimeoutSeconds(); + /** + * The maximum number of concurrent connections to the API host. + * + *

This value is used to configure both the Jetty/JDK HTTP client's connection pool limit + * and, when applicable, the maximum number of threads in the client's thread pool. + */ abstract OptionalInt maxConnectionsPerDestination(); /** For testing that we handle missing Content-Length correctly. */ @@ -123,6 +129,24 @@ Config config() { return config; } + static int getMaxThreads(Config config) { + int maxThreads = config.maxConnectionsPerDestination().orElse(100); + String maxThreadsEnv = System.getenv("APPENGINE_API_MAX_THREADS"); + if (maxThreadsEnv != null) { + try { + int envMaxThreads = Integer.parseInt(maxThreadsEnv); + logger.atInfo().log( + "Overriding API max threads to %d from environment variable.", envMaxThreads); + return envMaxThreads; + } catch (NumberFormatException e) { + logger.atWarning().withCause(e).log( + "Invalid value for APPENGINE_API_MAX_THREADS: %s, using default %d", + maxThreadsEnv, maxThreads); + } + } + return maxThreads; + } + static HttpApiHostClient create(String url, Config config) { if (System.getenv("APPENGINE_API_CALLS_USING_JDK_CLIENT") != null) { logger.atInfo().log("Using JDK HTTP client for API calls"); diff --git a/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/HttpApiHostClientFactory.java b/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/HttpApiHostClientFactory.java index 9bcab0d92..81656d318 100644 --- a/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/HttpApiHostClientFactory.java +++ b/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/HttpApiHostClientFactory.java @@ -20,20 +20,43 @@ import com.google.apphosting.runtime.anyrpc.APIHostClientInterface; import com.google.apphosting.runtime.http.HttpApiHostClient.Config; +import com.google.common.flogger.GoogleLogger; import com.google.common.net.HostAndPort; import java.util.OptionalInt; /** Makes instances of {@link HttpApiHostClient}. */ public class HttpApiHostClientFactory { + private static final GoogleLogger logger = GoogleLogger.forEnclosingClass(); private HttpApiHostClientFactory() {} /** * Creates a new HttpApiHostClient instance to talk to the HTTP-based API server on the given host * and port. This method is called reflectively from ApiHostClientFactory. + * + *

The maximum number of concurrent connections can be configured by setting the {@code + * APPENGINE_API_MAX_CONNECTIONS} environment variable to a positive integer. If set, this + * value overrides the {@code maxConcurrentRpcs} parameter. + * + * @param hostAndPort The host and port of the API server. + * @param maxConcurrentRpcs The default maximum number of concurrent RPCs, used if the + * environment variable is not set. + * @return A new {@link APIHostClientInterface} instance. */ public static APIHostClientInterface create( HostAndPort hostAndPort, OptionalInt maxConcurrentRpcs) { String url = "http://" + hostAndPort + REQUEST_ENDPOINT; + String maxConnectionsEnv = System.getenv("APPENGINE_API_MAX_CONNECTIONS"); + if (maxConnectionsEnv != null) { + try { + int maxConnections = Integer.parseInt(maxConnectionsEnv); + if (maxConnections > 0) { + maxConcurrentRpcs = OptionalInt.of(maxConnections); + } + } catch (NumberFormatException e) { + logger.atWarning().withCause(e).log( + "Failed to parse APPENGINE_API_MAX_CONNECTIONS: %s", maxConnectionsEnv); + } + } Config config = Config.builder().setMaxConnectionsPerDestination(maxConcurrentRpcs).build(); return HttpApiHostClient.create(url, config); } diff --git a/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/JdkHttpApiHostClient.java b/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/JdkHttpApiHostClient.java index cb84007e5..4e5d9021c 100644 --- a/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/JdkHttpApiHostClient.java +++ b/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/JdkHttpApiHostClient.java @@ -17,6 +17,7 @@ package com.google.apphosting.runtime.http; import static java.lang.Math.max; +import static java.util.concurrent.TimeUnit.SECONDS; import com.google.apphosting.base.protos.RuntimePb.APIResponse; import com.google.apphosting.runtime.anyrpc.AnyRpcCallback; @@ -27,19 +28,29 @@ import java.io.InputStream; import java.io.OutputStream; import java.io.UncheckedIOException; +import java.lang.reflect.Method; import java.net.HttpURLConnection; import java.net.MalformedURLException; import java.net.SocketTimeoutException; import java.net.URL; import java.util.concurrent.Executor; import java.util.concurrent.Executors; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.Semaphore; import java.util.concurrent.ThreadFactory; +import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.atomic.AtomicInteger; /** * An alternative API client that uses the JDK's built-in HTTP client. This is likely to be much * less performant than {@link JettyHttpApiHostClient} but should allow us to determine whether * communications problems we are seeing are due to the Jetty client. + * + *

By default, this client uses a bounded thread pool to execute API calls, with the maximum + * number of threads determined by configuration. If the system property {@code + * appengine.api.use.virtualthreads} is set to {@code true}, it will instead use virtual threads via + * {@link Executors#newVirtualThreadPerTaskExecutor()}, with in-flight concurrency throttled by a + * semaphore to prevent backend overload. */ class JdkHttpApiHostClient extends HttpApiHostClient { private static final GoogleLogger logger = GoogleLogger.forEnclosingClass(); @@ -50,24 +61,75 @@ class JdkHttpApiHostClient extends HttpApiHostClient { private final URL url; private final Executor executor; + private final Semaphore concurrencySemaphore; - private JdkHttpApiHostClient(Config config, URL url, Executor executor) { + private JdkHttpApiHostClient( + Config config, URL url, Executor executor, Semaphore concurrencySemaphore) { super(config); this.url = url; this.executor = executor; + this.concurrencySemaphore = concurrencySemaphore; } + /** + * Creates a {@link JdkHttpApiHostClient}. + * + *

If the system property {@code appengine.api.use.virtualthreads} is set to {@code true}, a + * virtual thread executor is used to run requests with concurrency throttled to {@code + * maxThreads}. Otherwise, a bounded {@link ThreadPoolExecutor} is created, with {@code + * maxThreads} derived from {@code config.maxConnectionsPerDestination()}. + * + * @param url The URL of the API host. + * @param config Configuration for the client, including connection limits. + * @return A new {@link JdkHttpApiHostClient}. + */ + @SuppressWarnings("AllowVirtualThreads") static JdkHttpApiHostClient create(String url, Config config) { try { - ThreadFactory factory = - runnable -> { - Thread t = new Thread(rootThreadGroup(), runnable); - t.setName("JdkHttp-" + threadCount.incrementAndGet()); - t.setDaemon(true); - return t; - }; - Executor executor = Executors.newCachedThreadPool(factory); - return new JdkHttpApiHostClient(config, new URL(url), executor); + Executor executor = null; + Semaphore concurrencySemaphore = null; + int maxThreads = getMaxThreads(config); + if (Boolean.getBoolean("appengine.api.use.virtualthreads")) { + try { + Method newVirtualThreadPerTaskExecutor = + Executors.class.getMethod("newVirtualThreadPerTaskExecutor"); + executor = (Executor) newVirtualThreadPerTaskExecutor.invoke(null); + concurrencySemaphore = new Semaphore(maxThreads); + logger.atInfo().log( + "Using virtual threads for JdkHttpApiHostClient with concurrency capped at %d.", + maxThreads); + } catch (ReflectiveOperationException e) { + logger.atInfo().log( + "appengine.api.use.virtualthreads is true, but virtual threads are not available on" + + " this JDK. Falling back to thread pool for JdkHttpApiHostClient."); + } + } + if (executor == null) { + ThreadFactory factory = + runnable -> { + Thread t = new Thread(rootThreadGroup(), runnable); + t.setName("JdkHttp-" + threadCount.incrementAndGet()); + t.setDaemon(true); + return t; + }; + /* + * Thread Pool Configuration & Bug Analysis: + * + * Similar to the JettyHttpApiHostClient, we explicitly bound the thread pool. + * We cap the threads at `maxConnectionsPerDestination` (which defaults to 100) + * instead of a hardcoded 200 to prevent severe memory pressure (Thread Stack sizes) + * on smaller AppEngine instance classes like F1 (256MB) or F2 (512MB). + * An unbounded thread pool allows a failing RPC to rapidly spin up thousands + * of threads under retry, which overwhelms the JVM and the internal Datastore + * Appserver connection, forcing it to respond with masking INTERNAL_ERROR fallbacks. + */ + ThreadPoolExecutor tpe = + new ThreadPoolExecutor( + maxThreads, maxThreads, 60L, SECONDS, new LinkedBlockingQueue<>(), factory); + tpe.allowCoreThreadTimeOut(true); + executor = tpe; + } + return new JdkHttpApiHostClient(config, new URL(url), executor, concurrencySemaphore); } catch (MalformedURLException e) { throw new UncheckedIOException(e); } @@ -82,6 +144,13 @@ private static ThreadGroup rootThreadGroup() { return group; } + /** + * Asynchronously sends an API request to the API host using a thread pool. + * + * @param requestBytes The serialized API request. + * @param context The context for the request, including deadline information. + * @param callback Callback to be invoked with the API response or failure. + */ @Override void send( byte[] requestBytes, @@ -94,6 +163,16 @@ private void doSend( byte[] requestBytes, HttpApiHostClient.Context context, AnyRpcCallback callback) { + if (concurrencySemaphore != null) { + try { + concurrencySemaphore.acquire(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + communicationFailure( + context, "Interrupted waiting for API client concurrency semaphore", callback, e); + return; + } + } try { HttpURLConnection connection = (HttpURLConnection) url.openConnection(); connection.setDoOutput(true); @@ -130,14 +209,28 @@ private void doSend( } catch (IOException e) { logger.atWarning().withCause(e).log("IOException"); communicationFailure(context, e.toString(), callback, e); + } finally { + if (concurrencySemaphore != null) { + concurrencySemaphore.release(); + } } } + /** + * This operation is not supported by JdkHttpApiHostClient. + * + * @throws UnsupportedOperationException always. + */ @Override public void enable() { throw new UnsupportedOperationException(); } + /** + * This operation is not supported by JdkHttpApiHostClient. + * + * @throws UnsupportedOperationException always. + */ @Override public void disable() { throw new UnsupportedOperationException(); diff --git a/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/JettyHttpApiHostClient.java b/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/JettyHttpApiHostClient.java index f55286bc6..8afb923d7 100644 --- a/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/JettyHttpApiHostClient.java +++ b/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/JettyHttpApiHostClient.java @@ -32,11 +32,8 @@ import java.nio.channels.ClosedSelectorException; import java.util.Arrays; import java.util.Map; -import java.util.concurrent.Executors; import java.util.concurrent.RejectedExecutionException; -import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeoutException; -import java.util.concurrent.atomic.AtomicInteger; import org.eclipse.jetty.client.BytesRequestContent; import org.eclipse.jetty.client.HttpClient; import org.eclipse.jetty.client.HttpResponseException; @@ -48,6 +45,7 @@ import org.eclipse.jetty.http.HttpHeader; import org.eclipse.jetty.http.HttpMethod; import org.eclipse.jetty.io.EofException; +import org.eclipse.jetty.util.thread.QueuedThreadPool; import org.eclipse.jetty.util.thread.ScheduledExecutorScheduler; import org.eclipse.jetty.util.thread.Scheduler; @@ -55,8 +53,6 @@ class JettyHttpApiHostClient extends HttpApiHostClient { private static final GoogleLogger logger = GoogleLogger.forEnclosingClass(); - private static final AtomicInteger threadCount = new AtomicInteger(); - private final String url; private final HttpClient httpClient; @@ -66,6 +62,17 @@ private JettyHttpApiHostClient(String url, HttpClient httpClient, Config config) this.httpClient = httpClient; } + /** + * Creates and starts a {@link JettyHttpApiHostClient}. + * + *

The {@code config.maxConnectionsPerDestination()} parameter is used to configure both + * {@link HttpClient#setMaxConnectionsPerDestination(int)} and the maximum number of threads in + * the {@link QueuedThreadPool}. + * + * @param url The URL of the API host. + * @param config Configuration for the client. + * @return A new, started {@link JettyHttpApiHostClient}. + */ static JettyHttpApiHostClient create(String url, Config config) { Preconditions.checkNotNull(url); HttpClient httpClient = new HttpClient(); @@ -86,20 +93,38 @@ static JettyHttpApiHostClient create(String url, Config config) { boolean daemon = false; Scheduler scheduler = new ScheduledExecutorScheduler(schedulerName, daemon, myLoader, myThreadGroup); - ThreadFactory factory = - runnable -> { - Thread t = new Thread(myThreadGroup, runnable); - t.setName("JettyHttpApiHostClient-" + threadCount.incrementAndGet()); - t.setDaemon(true); - return t; - }; - // By default HttpClient will use a QueuedThreadPool with minThreads=8 and maxThreads=200. - // 8 threads is probably too much for most apps, especially since asynchronous I/O means that - // 8 concurrent API requests probably don't need that many threads. It's also not clear - // what advantage we'd get from using a QueuedThreadPool with a smaller minThreads value, versus - // just one of the standard java.util.concurrent pools. Here we have minThreads=1, maxThreads=∞, - // and idleTime=60 seconds. maxThreads=200 and maxThreads=∞ are probably equivalent in practice. - httpClient.setExecutor(Executors.newCachedThreadPool(factory)); + /* + * Thread Pool Configuration & Bug Analysis: + * + * In previous versions of the runtime, an unbounded CachedThreadPool was used here: + * `httpClient.setExecutor(Executors.newCachedThreadPool(factory));` + * + * Under high load (e.g., when a customer's custom retry logic aggressively retries failing + * RPCs like `BeginTransaction`), an unbounded thread pool creates thousands of threads instantly. + * This leads to a system collapse: + * 1. JVM Overload: The Java container becomes severely memory and CPU constrained. + * 2. Appserver Flooded: The avalanche of concurrent requests from the Java container floods the + * C++ Appserver proxy. + * 3. Triggering the C++ Bug Mask: Under massive load, the C++ Appserver's gRPC calls to the + * Datastore fail with UNAVAILABLE or RESOURCE_EXHAUSTED errors. + * 4. The Response: Because these aren't standard application errors, the C++ code + * (DatastoreClientHelper::DoneImpl) masks them as `Error::INTERNAL_ERROR` and returns the + * message "Internal Datastore Error" to the Java client to prevent leaking internal + * infrastructure details. + * 5. The Java client throws DatastoreFailureException, triggering the customer's loop again. + * + * To prevent this "retry storm", we explicitly use a bounded QueuedThreadPool. + * We cap the threads at `maxConnectionsPerDestination` (which defaults to 100) + * instead of a hardcoded 200 to prevent severe memory pressure (Thread Stack sizes) + * on smaller AppEngine instance classes like F1 (256MB) or F2 (512MB). + * If the system experiences a spike, Jetty will safely queue the outgoing RPCs, preventing the + * JVM and the Appserver from being overwhelmed and eliminating the INTERNAL_ERROR fallback loop. + */ + int maxThreads = getMaxThreads(config); + QueuedThreadPool threadPool = new QueuedThreadPool(maxThreads, 10, 60000, null, myThreadGroup); + threadPool.setName("JettyHttpApiHostClient"); + threadPool.setDaemon(true); + httpClient.setExecutor(threadPool); httpClient.setScheduler(scheduler); config.maxConnectionsPerDestination().ifPresent(httpClient::setMaxConnectionsPerDestination); try { @@ -220,6 +245,13 @@ && config().treatClosedChannelAsCancellation()) { } } + /** + * Asynchronously sends an API request to the API host. + * + * @param requestBytes The serialized API request. + * @param context The context for the request, including deadline information. + * @param callback Callback to be invoked with the API response or failure. + */ @Override void send( byte[] requestBytes, @@ -260,6 +292,10 @@ void send( request.send(completeListener); } + /** + * Disables the client by stopping the underlying {@link HttpClient}. Subsequent calls to {@link + * #send} may fail until {@link #enable()} is called. + */ @Override public synchronized void disable() { try { @@ -271,6 +307,7 @@ public synchronized void disable() { } } + /** Enables the client by starting the underlying {@link HttpClient}. */ @Override public synchronized void enable() { try { diff --git a/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/jetty/JettyServletEngineAdapter.java b/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/jetty/JettyServletEngineAdapter.java index ac7a8a1fe..a914b2b1f 100644 --- a/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/jetty/JettyServletEngineAdapter.java +++ b/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/jetty/JettyServletEngineAdapter.java @@ -103,9 +103,11 @@ public void start(String serverInfo, ServletEngineAdapter.Config runtimeOptions) boolean isHttpConnectorMode = Boolean.getBoolean(HTTP_CONNECTOR_MODE); QueuedThreadPool threadPool = new QueuedThreadPool(MAX_THREAD_POOL_THREADS, MIN_THREAD_POOL_THREADS); - // Try to enable virtual threads if requested and on java21: + // Try to enable virtual threads if requested and on Java 21+: if (Boolean.getBoolean("appengine.use.virtualthreads") - && ("java21".equals(GAE_RUNTIME) || "java25".equals(GAE_RUNTIME))) { + && (Runtime.version().feature() >= 21 + || "java21".equals(GAE_RUNTIME) + || "java25".equals(GAE_RUNTIME))) { threadPool.setVirtualThreadsExecutor(VirtualThreads.getDefaultVirtualThreadsExecutor()); logger.atInfo().log("Configuring Appengine web server virtual threads."); } From d3ebd53eef3c31f098b5a3a5e42ecb81174e6078 Mon Sep 17 00:00:00 2001 From: Ludovic Champenois Date: Wed, 2 Sep 2026 15:50:54 -0700 Subject: [PATCH 2/3] Fix flaky SizeLimitHandlerTest by tolerating aborted connection on oversized payload When streaming payloads exceeding the maximum size under HTTP connector mode, the server may abort or reset the TCP connection before HTTP 413 headers are received by the client, resulting in a response status of 0. This change accepts status 0 when a request/response failure is present, eliminating test flakiness. --- .../runtime/jetty/SizeLimitHandlerTest.java | 31 ++++++++++++------- 1 file changed, 19 insertions(+), 12 deletions(-) diff --git a/runtime/test/src/test/java/com/google/apphosting/runtime/jetty/SizeLimitHandlerTest.java b/runtime/test/src/test/java/com/google/apphosting/runtime/jetty/SizeLimitHandlerTest.java index 344053abf..5dc8ba2d7 100644 --- a/runtime/test/src/test/java/com/google/apphosting/runtime/jetty/SizeLimitHandlerTest.java +++ b/runtime/test/src/test/java/com/google/apphosting/runtime/jetty/SizeLimitHandlerTest.java @@ -20,6 +20,7 @@ import static org.hamcrest.CoreMatchers.containsString; import static org.hamcrest.CoreMatchers.equalTo; import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.anyOf; import static org.hamcrest.Matchers.lessThan; import static org.hamcrest.Matchers.lessThanOrEqualTo; import static org.junit.Assert.assertNotNull; @@ -247,11 +248,13 @@ public void testRequestContentAboveMaxLength() throws Exception { .send(completionListener::complete); Result result = completionListener.get(5, TimeUnit.SECONDS); - assertThat(result.getResponse().getStatus(), equalTo(HttpStatus.PAYLOAD_TOO_LARGE_413)); - - // If the request was not aborted, then we expect to see the error message in the response body. - if (result.getResponseFailure() == null) { + if (result.getResponseFailure() == null && result.getRequestFailure() == null) { + assertThat(result.getResponse().getStatus(), equalTo(HttpStatus.PAYLOAD_TOO_LARGE_413)); assertThat(received.toString(), containsString("Request body is too large")); + } else { + assertThat( + result.getResponse().getStatus(), + anyOf(equalTo(HttpStatus.PAYLOAD_TOO_LARGE_413), equalTo(0))); } } @@ -305,11 +308,13 @@ public void testRequestContentAboveMaxLengthGzip() throws Exception { .send(completionListener::complete); Result result = completionListener.get(5, TimeUnit.SECONDS); - assertThat(result.getResponse().getStatus(), equalTo(HttpStatus.PAYLOAD_TOO_LARGE_413)); - - // If the request was not aborted, then we expect to see the error message in the response body. - if (result.getResponseFailure() == null) { + if (result.getResponseFailure() == null && result.getRequestFailure() == null) { + assertThat(result.getResponse().getStatus(), equalTo(HttpStatus.PAYLOAD_TOO_LARGE_413)); assertThat(received.toString(), containsString("Request body is too large")); + } else { + assertThat( + result.getResponse().getStatus(), + anyOf(equalTo(HttpStatus.PAYLOAD_TOO_LARGE_413), equalTo(0))); } } @@ -356,11 +361,13 @@ public void testRequestContentLengthHeader() throws Exception { Result result = completionListener.get(5, TimeUnit.SECONDS); Response response = result.getResponse(); - assertThat(response.getStatus(), equalTo(HttpStatus.PAYLOAD_TOO_LARGE_413)); - - // If the request was not aborted, then we expect to see the error message in the response body. - if (result.getResponseFailure() == null) { + if (result.getResponseFailure() == null && result.getRequestFailure() == null) { + assertThat(response.getStatus(), equalTo(HttpStatus.PAYLOAD_TOO_LARGE_413)); assertThat(received.toString(), containsString("Request body is too large")); + } else { + assertThat( + response.getStatus(), + anyOf(equalTo(HttpStatus.PAYLOAD_TOO_LARGE_413), equalTo(0))); } } From 1ee50156f1753eaf36d97df9a01994077e7f8d52 Mon Sep 17 00:00:00 2001 From: Ludovic Champenois Date: Thu, 3 Sep 2026 10:16:16 -0700 Subject: [PATCH 3/3] Address code review feedback on virtual threads and high throughput - Evict and rollback transaction on beginTransaction failure in DatastoreServiceImpl to prevent thread-local leaks and server-side zombie transactions. - Handle non-200 HTTP responses in JdkHttpApiHostClient with communicationFailure callback instead of hanging callers. - Safely handle chunked transfer encoding (unknown Content-Length) in JdkHttpApiHostClient using ByteStreams.limit and ByteStreams.toByteArray. - Catch Throwable in JdkHttpApiHostClient around HTTP requests to guarantee API callbacks fire under errors. - Guard maxThreads and minThreads in HttpApiHostClient and JettyHttpApiHostClient to prevent startup IllegalArgumentException when maxThreads < 10. - Configure Jetty 12 JettyServletEngineAdapter with Jetty's VirtualThreads.getDefaultVirtualThreadsExecutor() rather than a custom ForkJoinPool. - Ensure strict Runtime.version().feature() >= 21 checks across Jetty servlet engines. - Remove unused VirtualThreadSupport classes from api, api_dev, and runtime/impl. - Add virtual thread support to JettyHttpApiHostClient when appengine.api.use.virtualthreads is set on Java 21+. --- .../AsyncDatastoreServiceInternal.java | 3 + .../BaseAsyncDatastoreServiceImpl.java | 5 + .../api/datastore/DatastoreServiceImpl.java | 10 +- .../appengine/setup/VirtualThreadSupport.java | 93 ------------------- .../development/VirtualThreadSupport.java | 93 ------------------- .../utils/runtime/VirtualThreadSupport.java | 93 ------------------- .../jetty/JettyServletEngineAdapter.java | 36 +------ runtime/runtime_impl_jetty121/API_CLIENTS.md | 7 +- .../runtime/http/HttpApiHostClient.java | 15 ++- .../runtime/http/JdkHttpApiHostClient.java | 26 +++++- .../runtime/http/JettyHttpApiHostClient.java | 10 +- .../jetty/JettyServletEngineAdapter.java | 4 +- 12 files changed, 69 insertions(+), 326 deletions(-) delete mode 100644 api/src/main/java/com/google/appengine/setup/VirtualThreadSupport.java delete mode 100644 api_dev/src/main/java/com/google/appengine/tools/development/VirtualThreadSupport.java delete mode 100644 runtime/impl/src/main/java/com/google/apphosting/utils/runtime/VirtualThreadSupport.java diff --git a/api/src/main/java/com/google/appengine/api/datastore/AsyncDatastoreServiceInternal.java b/api/src/main/java/com/google/appengine/api/datastore/AsyncDatastoreServiceInternal.java index b8f453925..a02d2abea 100644 --- a/api/src/main/java/com/google/appengine/api/datastore/AsyncDatastoreServiceInternal.java +++ b/api/src/main/java/com/google/appengine/api/datastore/AsyncDatastoreServiceInternal.java @@ -26,4 +26,7 @@ interface AsyncDatastoreServiceInternal extends AsyncDatastoreService { /** See {@link DatastoreService#allocateIdRange(KeyRange)}. */ Future allocateIdRange(final KeyRange range); + + /** Returns the default transaction stack provider. */ + TransactionStack getDefaultTxnProvider(); } diff --git a/api/src/main/java/com/google/appengine/api/datastore/BaseAsyncDatastoreServiceImpl.java b/api/src/main/java/com/google/appengine/api/datastore/BaseAsyncDatastoreServiceImpl.java index d323acef8..f67331e7c 100644 --- a/api/src/main/java/com/google/appengine/api/datastore/BaseAsyncDatastoreServiceImpl.java +++ b/api/src/main/java/com/google/appengine/api/datastore/BaseAsyncDatastoreServiceImpl.java @@ -393,6 +393,11 @@ public Future beginTransaction(TransactionOptions options) { return new FutureHelper.FakeFuture(txn); } + @Override + public TransactionStack getDefaultTxnProvider() { + return defaultTxnProvider; + } + private Transaction createTransaction(TransactionOptions options, boolean isExplicit) { return new TransactionImpl( datastoreServiceConfig.getAppIdNamespace().getAppId(), diff --git a/api/src/main/java/com/google/appengine/api/datastore/DatastoreServiceImpl.java b/api/src/main/java/com/google/appengine/api/datastore/DatastoreServiceImpl.java index 1e0d0d413..a5a74d660 100644 --- a/api/src/main/java/com/google/appengine/api/datastore/DatastoreServiceImpl.java +++ b/api/src/main/java/com/google/appengine/api/datastore/DatastoreServiceImpl.java @@ -148,13 +148,21 @@ public Transaction beginTransaction(TransactionOptions options) { int retries = 0; long delay = BEGIN_TXN_RETRY_DELAY_MS; while (true) { + Transaction tx = null; try { - Transaction tx = quietGet(async.beginTransaction(options)); + tx = quietGet(async.beginTransaction(options)); tx.getId(); // Force handle resolution return tx; } catch (DatastoreFailureException | DatastoreTimeoutException | ApiProxy.RPCFailedException e) { + if (tx != null) { + try { + tx.rollbackAsync(); + } catch (Exception ignored) { + } + async.getDefaultTxnProvider().remove(tx); + } if (++retries > MAX_RETRIES) { throw e; } diff --git a/api/src/main/java/com/google/appengine/setup/VirtualThreadSupport.java b/api/src/main/java/com/google/appengine/setup/VirtualThreadSupport.java deleted file mode 100644 index 361187c27..000000000 --- a/api/src/main/java/com/google/appengine/setup/VirtualThreadSupport.java +++ /dev/null @@ -1,93 +0,0 @@ -/* - * Copyright 2022 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 - * - * https://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.appengine.setup; - -import java.lang.invoke.MethodHandle; -import java.lang.invoke.MethodHandles; -import java.lang.invoke.MethodType; - -/** - * Utility class to support virtual threads using reflection, enabling compatibility with JDK 17. - */ -public class VirtualThreadSupport { - private static final MethodHandle OF_VIRTUAL; - private static final MethodHandle UNSTARTED; - private static final MethodHandle IS_VIRTUAL; - - static { - MethodHandle ofVirtual = null; - MethodHandle unstarted = null; - MethodHandle isVirtual = null; - try { - MethodHandles.Lookup lookup = MethodHandles.publicLookup(); - Class threadClass = Thread.class; - // In JDK 21+, Thread.ofVirtual() returns a Thread.Builder.OfVirtual - Class builderClass = Class.forName("java.lang.Thread$Builder$OfVirtual"); - - ofVirtual = lookup.findStatic(threadClass, "ofVirtual", MethodType.methodType(builderClass)); - unstarted = lookup.findVirtual(builderClass, "unstarted", MethodType.methodType(threadClass, Runnable.class)); - isVirtual = lookup.findVirtual(threadClass, "isVirtual", MethodType.methodType(boolean.class)); - } catch (ClassNotFoundException | NoSuchMethodException | IllegalAccessException ignored) { - // Not on JDK 21+ or virtual threads not available - } - OF_VIRTUAL = ofVirtual; - UNSTARTED = unstarted; - IS_VIRTUAL = isVirtual; - } - - /** - * Returns true if virtual threads are supported by the current JVM. - */ - public static boolean isSupported() { - return OF_VIRTUAL != null; - } - - /** - * Creates an unstarted virtual thread if supported. - * - * @param runnable the runnable to execute - * @return an unstarted virtual thread, or null if not supported - */ - public static Thread createVirtualThread(Runnable runnable) { - if (OF_VIRTUAL == null || UNSTARTED == null) { - return null; - } - try { - Object builder = OF_VIRTUAL.invoke(); - return (Thread) UNSTARTED.invoke(builder, runnable); - } catch (Throwable t) { - return null; - } - } - - /** - * Checks if the given thread is a virtual thread. - * - * @param thread the thread to check - * @return true if the thread is virtual, false otherwise - */ - public static boolean isVirtual(Thread thread) { - if (IS_VIRTUAL == null) { - return false; - } - try { - return (boolean) IS_VIRTUAL.invoke(thread); - } catch (Throwable t) { - return false; - } - } -} diff --git a/api_dev/src/main/java/com/google/appengine/tools/development/VirtualThreadSupport.java b/api_dev/src/main/java/com/google/appengine/tools/development/VirtualThreadSupport.java deleted file mode 100644 index 79f65f35a..000000000 --- a/api_dev/src/main/java/com/google/appengine/tools/development/VirtualThreadSupport.java +++ /dev/null @@ -1,93 +0,0 @@ -/* - * Copyright 2021 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 - * - * https://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.appengine.tools.development; - -import java.lang.invoke.MethodHandle; -import java.lang.invoke.MethodHandles; -import java.lang.invoke.MethodType; - -/** - * Utility class to support virtual threads using reflection, enabling compatibility with JDK 17. - */ -public class VirtualThreadSupport { - private static final MethodHandle OF_VIRTUAL; - private static final MethodHandle UNSTARTED; - private static final MethodHandle IS_VIRTUAL; - - static { - MethodHandle ofVirtual = null; - MethodHandle unstarted = null; - MethodHandle isVirtual = null; - try { - MethodHandles.Lookup lookup = MethodHandles.publicLookup(); - Class threadClass = Thread.class; - // In JDK 21+, Thread.ofVirtual() returns a Thread.Builder.OfVirtual - Class builderClass = Class.forName("java.lang.Thread$Builder$OfVirtual"); - - ofVirtual = lookup.findStatic(threadClass, "ofVirtual", MethodType.methodType(builderClass)); - unstarted = lookup.findVirtual(builderClass, "unstarted", MethodType.methodType(threadClass, Runnable.class)); - isVirtual = lookup.findVirtual(threadClass, "isVirtual", MethodType.methodType(boolean.class)); - } catch (ClassNotFoundException | NoSuchMethodException | IllegalAccessException ignored) { - // Not on JDK 21+ or virtual threads not available - } - OF_VIRTUAL = ofVirtual; - UNSTARTED = unstarted; - IS_VIRTUAL = isVirtual; - } - - /** - * Returns true if virtual threads are supported by the current JVM. - */ - public static boolean isSupported() { - return OF_VIRTUAL != null; - } - - /** - * Creates an unstarted virtual thread if supported. - * - * @param runnable the runnable to execute - * @return an unstarted virtual thread, or null if not supported - */ - public static Thread createVirtualThread(Runnable runnable) { - if (OF_VIRTUAL == null || UNSTARTED == null) { - return null; - } - try { - Object builder = OF_VIRTUAL.invoke(); - return (Thread) UNSTARTED.invoke(builder, runnable); - } catch (Throwable t) { - return null; - } - } - - /** - * Checks if the given thread is a virtual thread. - * - * @param thread the thread to check - * @return true if the thread is virtual, false otherwise - */ - public static boolean isVirtual(Thread thread) { - if (IS_VIRTUAL == null) { - return false; - } - try { - return (boolean) IS_VIRTUAL.invoke(thread); - } catch (Throwable t) { - return false; - } - } -} diff --git a/runtime/impl/src/main/java/com/google/apphosting/utils/runtime/VirtualThreadSupport.java b/runtime/impl/src/main/java/com/google/apphosting/utils/runtime/VirtualThreadSupport.java deleted file mode 100644 index 2fc126e9f..000000000 --- a/runtime/impl/src/main/java/com/google/apphosting/utils/runtime/VirtualThreadSupport.java +++ /dev/null @@ -1,93 +0,0 @@ -/* - * Copyright 2021 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 - * - * https://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.apphosting.utils.runtime; - -import java.lang.invoke.MethodHandle; -import java.lang.invoke.MethodHandles; -import java.lang.invoke.MethodType; - -/** - * Utility class to support virtual threads using reflection, enabling compatibility with JDK 17. - */ -public class VirtualThreadSupport { - private static final MethodHandle OF_VIRTUAL; - private static final MethodHandle UNSTARTED; - private static final MethodHandle IS_VIRTUAL; - - static { - MethodHandle ofVirtual = null; - MethodHandle unstarted = null; - MethodHandle isVirtual = null; - try { - MethodHandles.Lookup lookup = MethodHandles.publicLookup(); - Class threadClass = Thread.class; - // In JDK 21+, Thread.ofVirtual() returns a Thread.Builder.OfVirtual - Class builderClass = Class.forName("java.lang.Thread$Builder$OfVirtual"); - - ofVirtual = lookup.findStatic(threadClass, "ofVirtual", MethodType.methodType(builderClass)); - unstarted = lookup.findVirtual(builderClass, "unstarted", MethodType.methodType(threadClass, Runnable.class)); - isVirtual = lookup.findVirtual(threadClass, "isVirtual", MethodType.methodType(boolean.class)); - } catch (ClassNotFoundException | NoSuchMethodException | IllegalAccessException ignored) { - // Not on JDK 21+ or virtual threads not available - } - OF_VIRTUAL = ofVirtual; - UNSTARTED = unstarted; - IS_VIRTUAL = isVirtual; - } - - /** - * Returns true if virtual threads are supported by the current JVM. - */ - public static boolean isSupported() { - return OF_VIRTUAL != null; - } - - /** - * Creates an unstarted virtual thread if supported. - * - * @param runnable the runnable to execute - * @return an unstarted virtual thread, or null if not supported - */ - public static Thread createVirtualThread(Runnable runnable) { - if (OF_VIRTUAL == null || UNSTARTED == null) { - return null; - } - try { - Object builder = OF_VIRTUAL.invoke(); - return (Thread) UNSTARTED.invoke(builder, runnable); - } catch (Throwable t) { - return null; - } - } - - /** - * Checks if the given thread is a virtual thread. - * - * @param thread the thread to check - * @return true if the thread is virtual, false otherwise - */ - public static boolean isVirtual(Thread thread) { - if (IS_VIRTUAL == null) { - return false; - } - try { - return (boolean) IS_VIRTUAL.invoke(thread); - } catch (Throwable t) { - return false; - } - } -} diff --git a/runtime/runtime_impl_jetty12/src/main/java/com/google/apphosting/runtime/jetty/JettyServletEngineAdapter.java b/runtime/runtime_impl_jetty12/src/main/java/com/google/apphosting/runtime/jetty/JettyServletEngineAdapter.java index 0daec0ee2..66bbf78b2 100644 --- a/runtime/runtime_impl_jetty12/src/main/java/com/google/apphosting/runtime/jetty/JettyServletEngineAdapter.java +++ b/runtime/runtime_impl_jetty12/src/main/java/com/google/apphosting/runtime/jetty/JettyServletEngineAdapter.java @@ -48,8 +48,6 @@ import java.io.InputStreamReader; import java.util.Objects; import java.util.concurrent.ExecutionException; -import java.util.concurrent.Executor; -import java.util.concurrent.ForkJoinPool; import org.eclipse.jetty.http.CookieCompliance; import org.eclipse.jetty.http.HttpCompliance; import org.eclipse.jetty.http.MultiPartCompliance; @@ -105,17 +103,9 @@ public void start(String serverInfo, ServletEngineAdapter.Config runtimeOptions) new QueuedThreadPool(MAX_THREAD_POOL_THREADS, MIN_THREAD_POOL_THREADS); // Try to enable virtual threads if requested and on Java 21+: if (Boolean.getBoolean("appengine.use.virtualthreads") - && (Runtime.version().feature() >= 21 - || "java21".equals(GAE_RUNTIME) - || "java25".equals(GAE_RUNTIME))) { - int maxParallelism = getMaxSafeCarrierParallelism(); - Executor virtualThreadsExecutor = - new ForkJoinPool( - maxParallelism, ForkJoinPool.defaultForkJoinWorkerThreadFactory, null, true); - threadPool.setVirtualThreadsExecutor(virtualThreadsExecutor); - logger.atInfo().log( - "Configuring Appengine web server virtual threads with capped carrier parallelism: %d", - maxParallelism); + && Runtime.version().feature() >= 21) { + threadPool.setVirtualThreadsExecutor(VirtualThreads.getDefaultVirtualThreadsExecutor()); + logger.atInfo().log("Configuring Appengine web server virtual threads."); } server = @@ -280,24 +270,4 @@ public void serviceRequest(UPRequest upRequest, MutableUpResponse upResponse) th upResponse.setErrorMessage("Unexpected Error: " + error); } } - - /** - * Calculates a safe maximum carrier thread count based on GAE sandbox memory boundaries to - * prevent OS scheduling thrashing on fractional/low-core instances. - */ - static int getMaxSafeCarrierParallelism() { - return getMaxSafeCarrierParallelism(System.getenv("GAE_MEMORY_MB")); - } - - static int getMaxSafeCarrierParallelism(String memoryMbStr) { - if (memoryMbStr == null || memoryMbStr.isEmpty()) { - return 4; // Conservative default cap for standard runtimes - } - try { - int memoryMb = Integer.parseInt(memoryMbStr); - return memoryMb <= 512 ? 1 : memoryMb <= 1024 ? 2 : 4; - } catch (NumberFormatException e) { - return 4; // Safety Fallback - } - } } diff --git a/runtime/runtime_impl_jetty121/API_CLIENTS.md b/runtime/runtime_impl_jetty121/API_CLIENTS.md index 48af0f192..7e9e63620 100644 --- a/runtime/runtime_impl_jetty121/API_CLIENTS.md +++ b/runtime/runtime_impl_jetty121/API_CLIENTS.md @@ -48,12 +48,17 @@ and system properties: * **`APPENGINE_API_MAX_THREADS`** (Environment Variable): Sets the maximum number of concurrent threads for executing API calls. If unset, this also defaults to 100. This is the most direct way to control API - call throughput and prevent backend overload. + call throughput and prevent backend overload. If set to lower values (e.g. 5 or 8), + `minThreads` automatically scales down (`min(10, maxThreads)`) to prevent startup failures. * Default: `100` * **`APPENGINE_API_CALLS_IDLE_TIMEOUT_MS`** (Environment Variable): Sets the idle timeout in milliseconds for connections in the connection pool. Connections that are idle for longer than this duration may be closed. * Default: `58000` (58 seconds) +* **`appengine.api.use.virtualthreads`** (Java System Property): If set to `true` on + Java 21+, the client will use Java Virtual Threads to execute API requests, avoiding + platform thread stack memory overhead while honoring pool and connection limits. + * Default: `false` ## JDK HTTP Client diff --git a/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/HttpApiHostClient.java b/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/HttpApiHostClient.java index d5426e36e..3fcf5173d 100644 --- a/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/HttpApiHostClient.java +++ b/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/HttpApiHostClient.java @@ -131,13 +131,22 @@ Config config() { static int getMaxThreads(Config config) { int maxThreads = config.maxConnectionsPerDestination().orElse(100); + if (maxThreads <= 0) { + maxThreads = 100; + } String maxThreadsEnv = System.getenv("APPENGINE_API_MAX_THREADS"); if (maxThreadsEnv != null) { try { int envMaxThreads = Integer.parseInt(maxThreadsEnv); - logger.atInfo().log( - "Overriding API max threads to %d from environment variable.", envMaxThreads); - return envMaxThreads; + if (envMaxThreads > 0) { + logger.atInfo().log( + "Overriding API max threads to %d from environment variable.", envMaxThreads); + return envMaxThreads; + } else { + logger.atWarning().log( + "APPENGINE_API_MAX_THREADS must be positive: %d, using default %d", + envMaxThreads, maxThreads); + } } catch (NumberFormatException e) { logger.atWarning().withCause(e).log( "Invalid value for APPENGINE_API_MAX_THREADS: %s, using default %d", diff --git a/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/JdkHttpApiHostClient.java b/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/JdkHttpApiHostClient.java index 4e5d9021c..6015f04f5 100644 --- a/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/JdkHttpApiHostClient.java +++ b/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/JdkHttpApiHostClient.java @@ -190,25 +190,41 @@ private void doSend( try (OutputStream out = connection.getOutputStream()) { out.write(requestBytes); } - if (connection.getResponseCode() == HttpURLConnection.HTTP_OK) { + int responseCode = connection.getResponseCode(); + if (responseCode == HttpURLConnection.HTTP_OK) { int length = connection.getContentLength(); if (length > MAX_LENGTH) { connection.getInputStream().close(); responseTooBig(callback); - } else { + } else if (length >= 0) { byte[] buffer = new byte[length]; try (InputStream in = connection.getInputStream()) { ByteStreams.readFully(in, buffer); // EOFException (an IOException) if too few bytes receivedResponse(buffer, length, context, callback); } + } else { + // Chunked transfer encoding or unspecified content length + byte[] buffer; + try (InputStream in = connection.getInputStream()) { + buffer = ByteStreams.toByteArray(ByteStreams.limit(in, MAX_LENGTH + 1)); + } + if (buffer.length > MAX_LENGTH) { + responseTooBig(callback); + } else { + receivedResponse(buffer, buffer.length, context, callback); + } } + } else { + String httpError = responseCode + " " + connection.getResponseMessage(); + logger.atWarning().log("HTTP communication got error: %s", httpError); + communicationFailure(context, httpError, callback, null); } } catch (SocketTimeoutException e) { logger.atWarning().withCause(e).log("SocketTimeoutException"); timeout(callback); - } catch (IOException e) { - logger.atWarning().withCause(e).log("IOException"); - communicationFailure(context, e.toString(), callback, e); + } catch (Throwable t) { + logger.atWarning().withCause(t).log("HTTP communication failure"); + communicationFailure(context, t.toString(), callback, t); } finally { if (concurrencySemaphore != null) { concurrencySemaphore.release(); diff --git a/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/JettyHttpApiHostClient.java b/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/JettyHttpApiHostClient.java index 8afb923d7..ff5bec685 100644 --- a/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/JettyHttpApiHostClient.java +++ b/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/JettyHttpApiHostClient.java @@ -45,6 +45,7 @@ import org.eclipse.jetty.http.HttpHeader; import org.eclipse.jetty.http.HttpMethod; import org.eclipse.jetty.io.EofException; +import org.eclipse.jetty.util.VirtualThreads; import org.eclipse.jetty.util.thread.QueuedThreadPool; import org.eclipse.jetty.util.thread.ScheduledExecutorScheduler; import org.eclipse.jetty.util.thread.Scheduler; @@ -121,9 +122,16 @@ static JettyHttpApiHostClient create(String url, Config config) { * JVM and the Appserver from being overwhelmed and eliminating the INTERNAL_ERROR fallback loop. */ int maxThreads = getMaxThreads(config); - QueuedThreadPool threadPool = new QueuedThreadPool(maxThreads, 10, 60000, null, myThreadGroup); + int minThreads = Math.max(1, Math.min(10, maxThreads)); + QueuedThreadPool threadPool = + new QueuedThreadPool(maxThreads, minThreads, 60000, null, myThreadGroup); threadPool.setName("JettyHttpApiHostClient"); threadPool.setDaemon(true); + if (Boolean.getBoolean("appengine.api.use.virtualthreads") + && Runtime.version().feature() >= 21) { + threadPool.setVirtualThreadsExecutor(VirtualThreads.getDefaultVirtualThreadsExecutor()); + logger.atInfo().log("Using virtual threads for JettyHttpApiHostClient."); + } httpClient.setExecutor(threadPool); httpClient.setScheduler(scheduler); config.maxConnectionsPerDestination().ifPresent(httpClient::setMaxConnectionsPerDestination); diff --git a/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/jetty/JettyServletEngineAdapter.java b/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/jetty/JettyServletEngineAdapter.java index a914b2b1f..960585f36 100644 --- a/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/jetty/JettyServletEngineAdapter.java +++ b/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/jetty/JettyServletEngineAdapter.java @@ -105,9 +105,7 @@ public void start(String serverInfo, ServletEngineAdapter.Config runtimeOptions) new QueuedThreadPool(MAX_THREAD_POOL_THREADS, MIN_THREAD_POOL_THREADS); // Try to enable virtual threads if requested and on Java 21+: if (Boolean.getBoolean("appengine.use.virtualthreads") - && (Runtime.version().feature() >= 21 - || "java21".equals(GAE_RUNTIME) - || "java25".equals(GAE_RUNTIME))) { + && Runtime.version().feature() >= 21) { threadPool.setVirtualThreadsExecutor(VirtualThreads.getDefaultVirtualThreadsExecutor()); logger.atInfo().log("Configuring Appengine web server virtual threads."); }