This is an automated email from the ASF dual-hosted git repository.
rzo1 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/storm.git
The following commit(s) were added to refs/heads/master by this push:
new d4db97d8d Add an opt-in virtual threads flag for the Thrift server
handler pools on Java 25 (#9097)
d4db97d8d is described below
commit d4db97d8d6484cc9868ae11e3bc75661d7662ff2
Author: Gianluca Graziadei <[email protected]>
AuthorDate: Sun Oct 25 19:32:48 2026 +0100
Add an opt-in virtual threads flag for the Thrift server handler pools on
Java 25 (#9097)
* Add opt-in virtual threads for blocking I/O thread pools
Introduce storm.virtual.threads.enabled (default false) and
StormThreadFactory, which returns a virtual-thread ThreadFactory when the
flag is on and a non-daemon, normal-priority platform-thread factory
otherwise. Thread names are <prefix>-<n> in both modes.
The factory is injected into the existing pools, so core sizes, queues,
rejection policy and error handling are unchanged and the *.threads
settings keep their meaning as a concurrency bound:
- Thrift server handlers for the SASL, TLS and simple transports
(Nimbus, Supervisor and DRPC). With the simple transport an executor is
now always supplied when the flag is on; without a configured queue
size it uses the same unbounded queue THsHaServer builds by default.
- AsyncLocalizer download and task executors.
- Nimbus AssignmentDistributionService and Supervisor heartbeat pools.
- DRPCSpout background executor.
- Worker shared executor exposed through TopologyContext, read from the
merged topology conf so a topology can override the flag.
Spout/bolt executor threads, worker transfer, JCQueue, Netty and timer
threads are deliberately left on platform threads.
Round-trip tests through NimbusClient assert that Thrift handlers run on
virtual threads when the flag is on and on platform threads by default;
unit tests cover the factory, the localizer executor and the worker
shared executor.
ThriftHandlerVirtualThreadsBench in storm-perf measures the flag against
an in-process Nimbus Thrift server whose handler simulates blocking I/O.
At equal *.threads the throughput is unchanged, while server-side OS
threads drop from *.threads to roughly the number of cores.
* Document the handler concurrency change for servers without a queue size
With the simple transport and virtual threads on, an executor of
*.threads handlers is always supplied, which raises the effective
concurrency from THsHaServer's 5. The DRPC invocations port has no queue
size key, so this always applies there (5 -> drpc.invocations.threads),
and it also applies to any server whose *.queue.size was explicitly
unset. Say so in the code comment and in the javadoc of
storm.virtual.threads.enabled instead of describing it as a rare case.
* Correct the virtual threads caveats and make the jstack action dump
virtual threads too
The javadoc advised raising the carrier scheduler parallelism for blocking
file I/O, but on JDK 25 the scheduler already compensates for that and
synchronized no longer pins. Describe what actually matters instead:
native frames still pin a carrier, virtual threads are not time sliced so
a CPU-heavy Nimbus call can delay cheap ones, and jstack and
Thread.getAllStackTraces() do not show virtual threads, which makes the
thread dumps Storm logs itself blind for these pools.
Have the jstack action of bin/flight.bash also write a jcmd
Thread.dump_to_file dump, which does include virtual threads.
* Mention that the DRPCSpout background executor is also topology controlled
DRPCSpout builds its thread factory from the conf passed to open(), so a
topology-level storm.virtual.threads.enabled affects that pool as well,
not only the worker shared executor.
* Warn when storm.virtual.threads.enabled is a string other than true or
false
Boolean.parseBoolean silently mapped values such as "yes" or "1" to
false, while only non-String types hit the warning. String values only
reach the factory through storm.options, since daemon and topology confs
are validated as boolean, so keep accepting strings but log the ones that
are neither true nor false.
* Bound the wait for the benchmark Thrift server to start
The loop waiting for isServing() spun forever if serve() threw. Capture a
failure from the serve thread and give up after 30 seconds or when the
thread exits without serving.
* Compute benchmark throughput over completed calls only
Throughput divided clients * calls by the elapsed time, so a client that
failed part way still counted all of its calls, while the latency
percentiles only used the calls that completed. Use the completed count
for both and report attempted, completed and failed clients separately.
* Default the benchmark to a single mode per JVM
The javadoc recommends running each mode in its own JVM, but the default
--mode was both, which runs them back to back sharing heap and JIT state.
Default to off, reject unknown modes, and describe both as a quick look
only.
* Cover the no-queue-size handler path and the TLS transport in the virtual
thread tests
The simple transport test meant to cover the path where no *.queue.size
is configured, but withServer starts from readStormConfig(), which
carries nimbus.queue.size from defaults.yaml, so both simple transport
tests ran the ArrayBlockingQueue branch. Unset the key explicitly so the
LinkedBlockingQueue branch, the one the DRPC invocations port always
takes, is exercised.
Add a round trip through a TLS ThriftServer asserting the handler runs
on a virtual thread, which covers TThreadPoolServer with the TLS plugin.
The client is a plain Thrift client over a trust-only SSL socket because
the test certificate has no subject alternative name.
* Test the worker shared executor through a real WorkerState
The previous tests called the static helper directly, so reverting
makeDefaultResources to read the daemon conf would still have passed.
Build a WorkerState with the flag only in the topology conf and take the
executor from getDefaultSharedResources(), which fails if the wiring
regresses. The helper no longer needs to be visible for testing.
* Answer Object methods on the benchmark handler proxy for the proxy itself
equals, hashCode and toString were forwarded to the captured in-flight
counter instead of the proxy.
* Keep the original AsyncLocalizer thread names
The localizer threads were renamed from "AsyncLocalizer Download Executor
- 0" to "AsyncLocalizer-Download-Executor-0", which breaks any external
log or metrics tooling matching the old literal. Have StormThreadFactory
use the given prefix verbatim so callers can reproduce an existing
format, and pass the original names for the localizer pools. The new
pools keep their <prefix>-<n> names.
* Derive the Thrift handler thread name prefix from ThriftConnectionType
The three transport plugins built the same name from type.name() by
hand. Expose it as getHandlerThreadNamePrefix() next to the other
per-type values.
* Accept a raw conf map in StormThreadFactory
AssignmentDistributionService.prepare receives a raw Map and needed an
unchecked cast to call the factory. Take Map<?, ?> instead, which only
needs get().
* Note that virtual threads are always daemon threads
The platform branch of StormThreadFactory sets daemon(false), while
Thread.Builder.OfVirtual has no such option. The Nimbus and Supervisor
shutdown hooks close the pools without joining their threads and rely on
the delayed force halt, so Storm itself does not depend on non-daemon
pool threads, but say so in the javadoc for anyone embedding the
daemons.
* Collapse the handler thread type tests in AuthTest into one parameterized
test
The four tests differed only in transport, extra conf and expected
result. One @ParameterizedTest with a case list keeps the no-queue-size,
queue-size, default and digest cases and the ReqContext principal check
for the authenticated transport.
* Benchmark the SASL digest transport too and report the server thread
estimate
The benchmark only covered the simple transport (THsHaServer), while the
SASL and TLS transports secure clusters run use TThreadPoolServer with a
thread per connection. Add --transport simple|digest; the digest mode
writes a throwaway JAAS file with a fixed user. Since every client holds
its connection, and with it a handler thread, for the whole run, digest
needs --threads >= --clients and the benchmark rejects anything else
instead of timing out half the handshakes.
Print the server-side platform thread estimate (peak minus before minus
client threads) that the PR description quoted but the tool did not
output, and show the first client failure when there is one.
* Throttle concurrent client connects in the benchmark
With hundreds of clients opening connections at the same instant the
server socket's default listen backlog overflows and part of the
connections are reset before the run starts, which showed up as failed
clients in the digest runs at 500 and 2000 clients. Connection setup is
not what is measured, so limit concurrent connects to 16; calls still
start only once every client is connected.
* Address minor changes
* Wrap new code to the 100 column checkstyle limit
Reflow the javadoc and comments added by this change and split the few
long code lines so the lines this PR introduces produce no checkstyle
warnings. In the benchmark, parse and validate each option next to its
first use and compute the effective queue size right before the report,
which also clears the VariableDeclarationUsageDistance warnings; add the
missing javadoc on main. No behaviour change.
* Limit storm.virtual.threads.enabled to the Thrift handler pools
The supervisor heartbeat, AsyncLocalizer, AssignmentDistributionService,
worker shared executor and DRPCSpout background pools go back to platform
threads. Their saving is a handful of threads, while on virtual threads
they would vanish from Utils.threadDump(), which Storm logs when a slot
gets stuck or before a forced halt.
flight.bash now writes jcmd's own output of the thread dump as -jcmd.txt:
the .log suffix made the log cleaner treat it as an active log and kept
it out of the logviewer dump list, which only shows .txt/.jfr/.bin.
---
bin/flight.bash | 9 +
conf/defaults.yaml | 1 +
.../perf/ThriftHandlerVirtualThreadsBench.java | 448 +++++++++++++++++++++
storm-client/src/jvm/org/apache/storm/Config.java | 41 ++
.../storm/security/auth/SimpleTransportPlugin.java | 19 +-
.../storm/security/auth/ThriftConnectionType.java | 9 +
.../security/auth/sasl/SaslTransportPlugin.java | 4 +-
.../security/auth/tls/TlsTransportPlugin.java | 4 +-
.../org/apache/storm/utils/StormThreadFactory.java | 101 +++++
.../security/auth/TlsTransportPluginTest.java | 66 ++-
.../apache/storm/utils/StormThreadFactoryTest.java | 106 +++++
.../org/apache/storm/security/auth/AuthTest.java | 62 ++-
12 files changed, 864 insertions(+), 6 deletions(-)
diff --git a/bin/flight.bash b/bin/flight.bash
index d38dd2e1d..0d6e1ed7b 100755
--- a/bin/flight.bash
+++ b/bin/flight.bash
@@ -98,6 +98,15 @@ function get_recording_timestamps {
function jstack_record {
FILENAME=jstack-$1-${NOW}.txt
${BINPATH}jstack $1 > "$2/${FILENAME}" 2>&1
+ # jstack does not list virtual threads (storm.virtual.threads.enabled);
Thread.dump_to_file does. The dump is
+ # written by the target JVM, so the directory is made absolute (the JVM
resolves relative paths against its own
+ # cwd) and jcmd's own output, including the reason when the dump fails, is
kept next to it as .txt so the
+ # logviewer lists it and the log cleaner does not treat it as an active
log.
+ if [ -x "${BINPATH}jcmd" ] || command -v "${BINPATH}jcmd" > /dev/null
2>&1; then
+ DUMPDIR=$(cd "$2" && pwd)
+ ${BINPATH}jcmd $1 Thread.dump_to_file -format=text
"${DUMPDIR}/threaddump-$1-${NOW}.txt" \
+ > "$2/threaddump-$1-${NOW}-jcmd.txt" 2>&1
+ fi
}
function jmap_record {
diff --git a/conf/defaults.yaml b/conf/defaults.yaml
index 219aff24d..3b49c151e 100644
--- a/conf/defaults.yaml
+++ b/conf/defaults.yaml
@@ -38,6 +38,7 @@ storm.zookeeper.auth.password: null
storm.zookeeper.ssl.enable: false
storm.zookeeper.ssl.hostnameVerification: false
storm.cluster.mode: "distributed" # can be distributed or local
+storm.virtual.threads.enabled: false
storm.local.mode.zmq: false
storm.thrift.transport: "org.apache.storm.security.auth.SimpleTransportPlugin"
storm.thrift.socket.timeout.ms: 600000
diff --git
a/examples/storm-perf/src/main/java/org/apache/storm/perf/ThriftHandlerVirtualThreadsBench.java
b/examples/storm-perf/src/main/java/org/apache/storm/perf/ThriftHandlerVirtualThreadsBench.java
new file mode 100644
index 000000000..684c44934
--- /dev/null
+++
b/examples/storm-perf/src/main/java/org/apache/storm/perf/ThriftHandlerVirtualThreadsBench.java
@@ -0,0 +1,448 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.storm.perf;
+
+import java.io.IOException;
+import java.lang.management.ManagementFactory;
+import java.lang.management.ThreadMXBean;
+import java.lang.reflect.InvocationHandler;
+import java.lang.reflect.Method;
+import java.lang.reflect.Proxy;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.storm.Config;
+import org.apache.storm.generated.Nimbus;
+import org.apache.storm.security.auth.SimpleTransportPlugin;
+import org.apache.storm.security.auth.ThriftConnectionType;
+import org.apache.storm.security.auth.ThriftServer;
+import org.apache.storm.security.auth.digest.DigestSaslTransportPlugin;
+import org.apache.storm.utils.NimbusClient;
+import org.apache.storm.utils.Utils;
+
+/**
+ * Benchmark for {@code storm.virtual.threads.enabled} on the Nimbus Thrift
handler pool.
+ *
+ * <p>Starts an in-process {@link ThriftServer} whose {@code getNimbusConf}
handler sleeps for a
+ * configurable time to simulate blocking I/O (ZooKeeper, blob store), then
hammers it with N client
+ * threads and reports throughput, latency percentiles, peak platform-thread
count, an estimate of
+ * the server-side platform threads and RSS. Run it with the flag off and on
to compare. The server
+ * can use either {@link SimpleTransportPlugin} ({@code THsHaServer}) or
+ * {@link DigestSaslTransportPlugin} ({@code TThreadPoolServer}, one handler
thread per connection,
+ * as used by the SASL and TLS transports of secure clusters); the digest mode
writes a throwaway
+ * JAAS file with a fixed user and needs {@code --threads >= --clients}, since
every client holds
+ * its connection, and with it a handler thread, for the whole run.
+ *
+ * <p>Build and run (JDK 25, Maven 3.9):
+ * <pre>
+ * mvn -pl storm-client,examples/storm-perf -am install -DskipTests
-Dcheckstyle.skip=true -q
+ * mvn -pl examples/storm-perf dependency:build-classpath
-Dmdep.outputFile=target/cp.txt -q
+ * java -cp examples/storm-perf/target/classes:$(cat
examples/storm-perf/target/cp.txt) \
+ * org.apache.storm.perf.ThriftHandlerVirtualThreadsBench \
+ * --mode off --clients 200 --calls 50 --io-ms 10
+ * java -cp examples/storm-perf/target/classes:$(cat
examples/storm-perf/target/cp.txt) \
+ * org.apache.storm.perf.ThriftHandlerVirtualThreadsBench \
+ * --mode on --clients 200 --calls 50 --io-ms 10
+ * </pre>
+ *
+ * <p>Options: {@code --mode off|on|both} (default {@code off}; {@code both}
runs the two modes in
+ * one JVM, which shares heap and JIT state between them and is only meant for
a quick look),
+ * {@code --clients N}, {@code --calls M} (per client), {@code --transport
simple|digest} (default
+ * {@code simple}), {@code --io-ms L} (simulated handler I/O), {@code
--threads T}
+ * ({@code nimbus.thrift.threads}), {@code --queue-size Q|none} ({@code
nimbus.queue.size};
+ * defaults to the shipped 100000, so both modes use a pool of {@code
--threads} handlers;
+ * {@code none} removes the key so the platform-thread run falls back to
THsHaServer's own pool,
+ * whose effective concurrency is 5), {@code --warmup W} (calls per client
excluded from
+ * statistics). The report and its JSON line carry the queue size the server
actually ran with
+ * ({@code null} only for {@code none}).
+ *
+ * <p>For comparable numbers run each mode in its own JVM ({@code --mode off}
then
+ * {@code --mode on}).
+ */
+public final class ThriftHandlerVirtualThreadsBench {
+
+ private static final String HOST = "localhost";
+ private static final int CONCURRENT_CONNECTS = 16;
+
+ private ThriftHandlerVirtualThreadsBench() {
+ }
+
+ /**
+ * Parses the options described in the class javadoc and runs the
requested mode(s).
+ *
+ * @param args {@code --name value} pairs
+ */
+ public static void main(String[] args) throws Exception {
+ Map<String, String> opts = parseArgs(args);
+ String mode = opts.getOrDefault("mode",
"off").toLowerCase(Locale.ROOT);
+ if (!mode.equals("off") && !mode.equals("on") && !mode.equals("both"))
{
+ throw new IllegalArgumentException(
+ "Unknown --mode " + mode + ", expected off, on or both");
+ }
+ String transport = opts.getOrDefault("transport",
"simple").toLowerCase(Locale.ROOT);
+ if (!transport.equals("simple") && !transport.equals("digest")) {
+ throw new IllegalArgumentException(
+ "Unknown --transport " + transport + ", expected simple or
digest");
+ }
+ int clients = Integer.parseInt(opts.getOrDefault("clients", "200"));
+ int threads = Integer.parseInt(opts.getOrDefault("threads", "64"));
+ if (transport.equals("digest") && clients > threads) {
+ // TThreadPoolServer dedicates a handler thread to each connection
for its lifetime,
+ // and every client keeps its connection open for the whole run,
so connections beyond
+ // --threads would sit in the executor queue until their SASL
handshake times out.
+ throw new IllegalArgumentException(
+ "--transport digest needs --threads >= --clients (one handler
thread per open"
+ + " connection); got threads=" + threads + " clients=" +
clients);
+ }
+ long ioMs = Long.parseLong(opts.getOrDefault("io-ms", "10"));
+ if (ioMs < 0) {
+ throw new IllegalArgumentException("--io-ms must be >= 0, got " +
ioMs);
+ }
+ String queueOpt = opts.get("queue-size");
+ boolean noQueue = "none".equalsIgnoreCase(queueOpt);
+ Integer queueSize = queueOpt == null || noQueue ? null :
Integer.valueOf(queueOpt);
+ int calls = Integer.parseInt(opts.getOrDefault("calls", "50"));
+ int warmup = Integer.parseInt(opts.getOrDefault("warmup", "5"));
+ System.out.printf(Locale.ROOT,
+ "transport=%s clients=%d calls=%d io-ms=%d threads=%d
queue-size=%s warmup=%d%n",
+ transport, clients, calls, ioMs, threads,
queueSizeLabel(queueSize, noQueue), warmup);
+ if (mode.equals("off") || mode.equals("both")) {
+ runPhase(false, transport, clients, calls, ioMs, threads,
queueSize, noQueue, warmup);
+ }
+ if (mode.equals("on") || mode.equals("both")) {
+ runPhase(true, transport, clients, calls, ioMs, threads,
queueSize, noQueue, warmup);
+ }
+ }
+
+ private static void runPhase(boolean virtual, String transport, int
clients, int calls,
+ long ioMs, int threads, Integer queueSize,
boolean noQueue,
+ int warmup) throws Exception {
+ Map<String, Object> conf = new HashMap<>(Utils.readDefaultConfig());
+ if (transport.equals("digest")) {
+ conf.put(Config.STORM_THRIFT_TRANSPORT_PLUGIN,
+ DigestSaslTransportPlugin.class.getName());
+ conf.put("java.security.auth.login.config",
writeDigestJaasConf().toString());
+ } else {
+ conf.put(Config.STORM_THRIFT_TRANSPORT_PLUGIN,
SimpleTransportPlugin.class.getName());
+ }
+ conf.put(Config.NIMBUS_THRIFT_PORT, 0);
+ conf.put(Config.NIMBUS_THRIFT_THREADS, threads);
+ conf.put(Config.STORM_VIRTUAL_THREADS_ENABLED, virtual);
+ conf.put(Config.STORM_NIMBUS_RETRY_TIMES, 0);
+ if (noQueue) {
+ conf.remove(Config.NIMBUS_QUEUE_SIZE);
+ } else if (queueSize != null) {
+ conf.put(Config.NIMBUS_QUEUE_SIZE, queueSize);
+ }
+
+ AtomicInteger inFlight = new AtomicInteger();
+ AtomicInteger peakInFlight = new AtomicInteger();
+ Nimbus.Iface handler = sleepingHandler(ioMs, inFlight, peakInFlight);
+ ThriftServer server = new ThriftServer(conf, new
Nimbus.Processor<>(handler),
+ ThriftConnectionType.NIMBUS);
+ // Note: a transport failure inside serve() is handled by ThriftServer
itself, which logs
+ // it and halts the JVM with exit code 1
(ThriftServer.handleServerException), so it never
+ // reaches serveFailure; only an Error thrown by serve() does. A
silent exit code 1 before
+ // any result line means the server failed to start (bad JAAS file,
unbindable port, ...).
+ AtomicReference<Throwable> serveFailure = new AtomicReference<>();
+ Thread serveThread = new Thread(() -> {
+ try {
+ server.serve();
+ } catch (Throwable t) {
+ serveFailure.set(t);
+ }
+ }, "bench-thrift-serve");
+ serveThread.setDaemon(true);
+ serveThread.start();
+ long serveDeadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(30);
+ while (!server.isServing()) {
+ if (serveFailure.get() != null) {
+ throw new IllegalStateException("Thrift server failed to
start",
+ serveFailure.get());
+ }
+ if (!serveThread.isAlive()) {
+ throw new IllegalStateException("Thrift server thread exited
before serving");
+ }
+ if (System.nanoTime() > serveDeadline) {
+ throw new IllegalStateException("Thrift server did not start
serving within 30s");
+ }
+ Thread.sleep(10);
+ }
+ int port = server.getPort();
+
+ ThreadMXBean threadMx = ManagementFactory.getThreadMXBean();
+ // Sampled before the clients start and only reported at the end.
+ final long rssBeforeKb = rssKb();
+ int threadsBefore = threadMx.getThreadCount();
+ AtomicInteger peakThreads = new AtomicInteger(threadsBefore);
+ Thread sampler = new Thread(() -> {
+ while (!Thread.currentThread().isInterrupted()) {
+ peakThreads.accumulateAndGet(threadMx.getThreadCount(),
Math::max);
+ try {
+ Thread.sleep(20);
+ } catch (InterruptedException e) {
+ return;
+ }
+ }
+ }, "bench-thread-sampler");
+ sampler.setDaemon(true);
+ sampler.start();
+
+ long[][] latenciesNs = new long[clients][calls];
+ List<Throwable> failures = new ArrayList<>();
+ CountDownLatch ready = new CountDownLatch(clients);
+ CountDownLatch go = new CountDownLatch(1);
+ CountDownLatch done = new CountDownLatch(clients);
+ // The server socket uses the default listen backlog, so with hundreds
of clients opening
+ // at once part of the connections would be reset before the run
starts. Connection setup
+ // is not what is measured, so limit how many clients connect
concurrently; calls are only
+ // issued once every client is connected.
+ Semaphore connecting = new Semaphore(CONCURRENT_CONNECTS);
+ for (int c = 0; c < clients; c++) {
+ final int clientIdx = c;
+ Thread t = new Thread(() -> {
+ try (NimbusClient client = connect(conf, port, connecting)) {
+ Nimbus.Iface iface = client.getClient();
+ for (int i = 0; i < warmup; i++) {
+ iface.getNimbusConf();
+ }
+ ready.countDown();
+ go.await();
+ for (int i = 0; i < calls; i++) {
+ long start = System.nanoTime();
+ iface.getNimbusConf();
+ latenciesNs[clientIdx][i] = System.nanoTime() - start;
+ }
+ } catch (Throwable e) {
+ synchronized (failures) {
+ failures.add(e);
+ }
+ ready.countDown();
+ } finally {
+ done.countDown();
+ }
+ }, "bench-client-" + c);
+ t.start();
+ }
+ ready.await();
+ long startNs = System.nanoTime();
+ go.countDown();
+ done.await();
+ final long elapsedNs = System.nanoTime() - startNs;
+
+ sampler.interrupt();
+ sampler.join(1000);
+ final long rssAfterKb = rssKb();
+ server.stop();
+ serveThread.join(5000);
+ if (!failures.isEmpty()) {
+ System.err.println("First client failure (" + failures.size() + "
in total):");
+ failures.get(0).printStackTrace();
+ }
+
+ // What the server actually ran with: the explicit value, the shipped
default, or none.
+ Integer effectiveQueueSize =
ThriftConnectionType.NIMBUS.getQueueSize(conf);
+ report(virtual, transport, clients, calls, ioMs, threads,
effectiveQueueSize, latenciesNs,
+ elapsedNs, failures.size(), threadsBefore, peakThreads.get(),
peakInFlight.get(),
+ rssBeforeKb, rssAfterKb);
+ }
+
+ private static String queueSizeLabel(Integer queueSize, boolean noQueue) {
+ return noQueue ? "none" : queueSize == null ? "default" :
queueSize.toString();
+ }
+
+ private static void report(boolean virtual, String transport, int clients,
int calls,
+ long ioMs, int threads, Integer queueSize,
long[][] latenciesNs,
+ long elapsedNs, int failures, int
threadsBefore, int peakThreads,
+ int peakInFlight, long rssBeforeKb, long
rssAfterKb) {
+ long[] all = Arrays.stream(latenciesNs).flatMapToLong(Arrays::stream)
+ .filter(v -> v > 0).sorted().toArray();
+ int attempted = clients * calls;
+ int completed = all.length;
+ double elapsedS = elapsedNs / 1e9;
+ // Throughput and latency percentiles are both computed over the calls
that completed.
+ double throughput = completed / elapsedS;
+ double p50 = pct(all, 0.50);
+ double p90 = pct(all, 0.90);
+ double p99 = pct(all, 0.99);
+ double max = all.length == 0 ? 0 : all[all.length - 1] / 1e6;
+ // Handler concurrency: the configured pool with virtual threads or a
queue size (simple
+ // transport), or with TThreadPoolServer (digest); the platform-thread
simple transport
+ // without a queue size falls back to THsHaServer's own pool, whose
effective concurrency
+ // is 5 (see SimpleTransportPlugin).
+ boolean configuredPool = virtual || queueSize != null ||
!transport.equals("simple");
+ int concurrent = Math.min(clients, configuredPool ? threads : 5);
+ // With io-ms 0 there is no I/O floor and the ideal is unbounded, so
it is not reported.
+ String ideal = ioMs == 0
+ ? "n/a"
+ : String.format(Locale.ROOT, "%.0f", concurrent * 1000.0 / ioMs);
+ // Platform threads created after the "before" sample are the client
threads plus whatever
+ // the server added (handlers, acceptor/selector threads). Subtracting
the clients gives
+ // the server-side estimate.
+ int serverThreadsEstimate = Math.max(0, peakThreads - threadsBefore -
clients);
+
+ String label = (virtual ? "virtual threads ON" : "virtual threads OFF")
+ + ", " + transport + " transport";
+ System.out.println();
+ System.out.println("=== " + label + " ===");
+ System.out.printf(Locale.ROOT,
+ " calls : %d completed of %d attempted (%d clients
failed)%n",
+ completed, attempted, failures);
+ System.out.printf(Locale.ROOT, " elapsed : %.2f s%n",
elapsedS);
+ System.out.printf(Locale.ROOT,
+ " throughput : %.0f calls/s (ideal at %d concurrent
handlers: %s)%n",
+ throughput, concurrent, ideal);
+ System.out.printf(Locale.ROOT,
+ " latency p50/p90/p99: %.1f / %.1f / %.1f ms (max %.1f, floor
%d)%n",
+ p50, p90, p99, max, ioMs);
+ System.out.printf(Locale.ROOT, " peak handlers busy : %d%n",
peakInFlight);
+ System.out.printf(Locale.ROOT, " platform threads : %d before, %d
peak (+%d)%n",
+ threadsBefore, peakThreads, peakThreads - threadsBefore);
+ System.out.printf(Locale.ROOT,
+ " server threads : ~%d (peak minus before minus %d client
threads)%n",
+ serverThreadsEstimate, clients);
+ System.out.printf(Locale.ROOT, " RSS : %d MB before,
%d MB after%n",
+ rssBeforeKb / 1024, rssAfterKb / 1024);
+ System.out.printf(Locale.ROOT,
+ "JSON
{\"virtual\":%b,\"transport\":\"%s\",\"clients\":%d,\"calls\":%d,\"ioMs\":%d,"
+ + "\"threads\":%d,\"queueSize\":%s,"
+ +
"\"throughput\":%.1f,\"p50Ms\":%.2f,\"p90Ms\":%.2f,\"p99Ms\":%.2f,\"maxMs\":%.2f,"
+ +
"\"peakHandlersBusy\":%d,\"platformThreadsPeak\":%d,\"platformThreadsBefore\":%d,"
+ + "\"serverThreadsEstimate\":%d,"
+ +
"\"rssBeforeMb\":%d,\"rssAfterMb\":%d,\"completedCalls\":%d,\"failedClients\":%d}%n",
+ virtual, transport, clients, calls, ioMs, threads, queueSize,
throughput,
+ p50, p90, p99, max, peakInFlight, peakThreads, threadsBefore,
serverThreadsEstimate,
+ rssBeforeKb / 1024, rssAfterKb / 1024, completed, failures);
+ }
+
+ private static NimbusClient connect(Map<String, Object> conf, int port,
Semaphore connecting)
+ throws Exception {
+ connecting.acquire();
+ try {
+ return NimbusClient.Builder.withConf(conf).withTimeout(60_000)
+ .buildWithNimbusHostPort(HOST, port);
+ } finally {
+ connecting.release();
+ }
+ }
+
+ private static double pct(long[] sorted, double q) {
+ if (sorted.length == 0) {
+ return 0;
+ }
+ int idx = (int) Math.min(sorted.length - 1, Math.round(q *
(sorted.length - 1)));
+ return sorted[idx] / 1e6;
+ }
+
+ /**
+ * A Nimbus handler whose {@code getNimbusConf} blocks for {@code ioMs};
every other method is
+ * unsupported.
+ */
+ private static Nimbus.Iface sleepingHandler(long ioMs, AtomicInteger
inFlight,
+ AtomicInteger peakInFlight) {
+ InvocationHandler h = (Object proxy, Method method, Object[] margs) ->
{
+ if (method.getName().equals("getNimbusConf")) {
+ int now = inFlight.incrementAndGet();
+ peakInFlight.accumulateAndGet(now, Math::max);
+ try {
+ Thread.sleep(ioMs);
+ } finally {
+ inFlight.decrementAndGet();
+ }
+ return "{}";
+ }
+ if (method.getDeclaringClass() == Object.class) {
+ switch (method.getName()) {
+ case "hashCode":
+ return System.identityHashCode(proxy);
+ case "equals":
+ return proxy == margs[0];
+ case "toString":
+ return "sleepingHandler(" + ioMs + "ms)";
+ default:
+ throw new
UnsupportedOperationException(method.getName());
+ }
+ }
+ throw new UnsupportedOperationException(method.getName());
+ };
+ return (Nimbus.Iface)
Proxy.newProxyInstance(Nimbus.Iface.class.getClassLoader(),
+ new Class<?>[]{ Nimbus.Iface.class }, h);
+ }
+
+ /**
+ * Writes a JAAS configuration with a fixed digest user for the server and
client sides. The
+ * login module class is only named, never instantiated: Storm reads the
user and password from
+ * the entry options.
+ */
+ private static Path writeDigestJaasConf() throws IOException {
+ Path file = Files.createTempFile("storm-bench-jaas", ".conf");
+ file.toFile().deleteOnExit();
+ Files.writeString(file,
+ "StormServer {\n"
+ + " org.apache.zookeeper.server.auth.DigestLoginModule
required\n"
+ + " user_bench=\"benchsecret\";\n"
+ + "};\n"
+ + "StormClient {\n"
+ + " org.apache.zookeeper.server.auth.DigestLoginModule
required\n"
+ + " username=\"bench\"\n"
+ + " password=\"benchsecret\";\n"
+ + "};\n");
+ return file;
+ }
+
+ /** Resident set size from /proc/self/status, or -1 when unavailable. */
+ private static long rssKb() {
+ try {
+ for (String line :
Files.readAllLines(Paths.get("/proc/self/status"))) {
+ if (line.startsWith("VmRSS:")) {
+ return Long.parseLong(line.replaceAll("[^0-9]", ""));
+ }
+ }
+ } catch (IOException | RuntimeException e) {
+ // not Linux, or unreadable
+ }
+ return -1;
+ }
+
+ private static Map<String, String> parseArgs(String[] args) {
+ Map<String, String> opts = new HashMap<>();
+ for (int i = 0; i < args.length; i++) {
+ if (!args[i].startsWith("--")) {
+ throw new IllegalArgumentException("Unexpected argument: " +
args[i]);
+ }
+ String key = args[i].substring(2);
+ if (i + 1 >= args.length || args[i + 1].startsWith("--")) {
+ throw new IllegalArgumentException("Missing value for --" +
key);
+ }
+ opts.put(key, args[++i]);
+ }
+ return opts;
+ }
+}
diff --git a/storm-client/src/jvm/org/apache/storm/Config.java
b/storm-client/src/jvm/org/apache/storm/Config.java
index d24b7f99c..017ea100a 100644
--- a/storm-client/src/jvm/org/apache/storm/Config.java
+++ b/storm-client/src/jvm/org/apache/storm/Config.java
@@ -1081,6 +1081,47 @@ public class Config extends HashMap<String, Object> {
@IsInteger
public static final String TOPOLOGY_WORKER_SHARED_THREAD_POOL_SIZE =
"topology.worker.shared.thread.pool.size";
+ /**
+ * When true, the Thrift server handler pools run on Java virtual threads
instead of platform
+ * threads. This covers the Nimbus, Supervisor and DRPC servers with every
transport plugin
+ * (SASL, TLS and simple). Pool sizes configured through the corresponding
{@code *.threads}
+ * settings keep their meaning as a bound on concurrency. No other pool is
affected: the
+ * supervisor localizer and heartbeat executors, the Nimbus assignment
distribution service,
+ * worker and executor threads and Netty event loops always use platform
threads, so they stay
+ * visible in the thread dumps Storm logs itself.
+ *
+ * <p>DRPC server handler pools ({@code drpc.worker.threads}, {@code
drpc.invocations.threads})
+ * are included, since the DRPC server uses the same transport plugins.
+ *
+ * <p>With the simple (non-authenticated) Thrift transport, Storm builds
its own handler pool of
+ * {@code *.threads} threads whenever a {@code *.queue.size} is
configured, which the shipped
+ * defaults do for Nimbus, Supervisor and the DRPC client port. When no
queue size is configured
+ * {@code THsHaServer} falls back to its own pool, whose effective
concurrency is 5; enabling
+ * this flag in that case always supplies a pool of {@code *.threads}
threads, so the handler
+ * concurrency rises from 5 to that value. This is unconditionally the
case for the DRPC
+ * invocations port, which has no queue size setting, so its concurrency
goes from 5 to
+ * {@code drpc.invocations.threads} (64 by default). It also applies to
any Nimbus, Supervisor
+ * or DRPC server whose {@code *.queue.size} was explicitly unset.
+ *
+ * <p>Operational caveats. Virtual threads are always daemon threads,
while the platform threads
+ * these pools use otherwise are not; Storm's daemons exit explicitly and
do not depend on that,
+ * but an application embedding them (for example a local cluster) might.
All virtual threads in
+ * a JVM share one carrier scheduler sized to the number of available
processors. The scheduler
+ * compensates for blocking file I/O by temporarily adding carriers (up to
+ * {@code jdk.virtualThreadScheduler.maxPoolSize}), and since JDK 24
{@code synchronized} no
+ * longer pins a carrier; what still pins is a native frame on the stack,
for example Hadoop
+ * native libraries used by a blob store.
+ *
+ * <p>Diagnostics. {@code jstack} and {@code Thread.getAllStackTraces()}
do not report virtual
+ * threads, so with this flag on the thread dumps Storm logs itself (the
stuck-slot dump in the
+ * supervisor and the dump before a forced halt in a shutdown hook) do not
show the Thrift
+ * handlers, and a hung handler is simply absent from them. Use
+ * {@code jcmd <pid> Thread.dump_to_file -format=text <file>} instead;
+ * {@code bin/flight.bash <pid> jstack <dir>} writes such a dump next to
the jstack output when
+ * {@code jcmd} is available.
+ */
+ @IsBoolean
+ public static final String STORM_VIRTUAL_THREADS_ENABLED =
"storm.virtual.threads.enabled";
/**
* The interval in seconds to use for determining whether to throttle
error reported to
* Zookeeper. For example, an interval of 10
diff --git
a/storm-client/src/jvm/org/apache/storm/security/auth/SimpleTransportPlugin.java
b/storm-client/src/jvm/org/apache/storm/security/auth/SimpleTransportPlugin.java
index bea4deeca..8d1374fbd 100644
---
a/storm-client/src/jvm/org/apache/storm/security/auth/SimpleTransportPlugin.java
+++
b/storm-client/src/jvm/org/apache/storm/security/auth/SimpleTransportPlugin.java
@@ -20,6 +20,8 @@ import java.security.Principal;
import java.util.HashSet;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import javax.security.auth.Subject;
@@ -35,6 +37,7 @@ import org.apache.storm.thrift.transport.TSocket;
import org.apache.storm.thrift.transport.TTransport;
import org.apache.storm.thrift.transport.TTransportException;
import org.apache.storm.thrift.transport.layered.TFramedTransport;
+import org.apache.storm.utils.StormThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -74,9 +77,21 @@ public class SimpleTransportPlugin implements
ITransportPlugin {
serverArgs.maxReadBufferBytes = maxBufferSize;
- if (queueSize != null) {
+ if (queueSize != null ||
StormThreadFactory.isVirtualEnabled(topoConf)) {
+ // THsHaServer builds its own platform-thread pool (core 5,
unbounded
+ // LinkedBlockingQueue) when no executor is supplied, so when
virtual threads are
+ // requested we always supply an executor; without a configured
queue size we use the
+ // same unbounded queue so no request is rejected that would not
be rejected today. Note
+ // that this raises the effective handler concurrency from
THsHaServer's 5 to
+ // numWorkerThreads. That is always the case for DRPC_INVOCATIONS,
which has no queue
+ // size key at all, and for any Nimbus, Supervisor or DRPC server
whose *.queue.size was
+ // explicitly unset.
+ BlockingQueue<Runnable> workQueue = queueSize != null
+ ? new ArrayBlockingQueue<>(queueSize)
+ : new LinkedBlockingQueue<>();
serverArgs.executorService(new
ThreadPoolExecutor(numWorkerThreads, numWorkerThreads,
- 60, TimeUnit.SECONDS, new ArrayBlockingQueue(queueSize)));
+ 60, TimeUnit.SECONDS, workQueue,
+ StormThreadFactory.create(topoConf,
type.getHandlerThreadNamePrefix())));
}
// construct THsHaServer
diff --git
a/storm-client/src/jvm/org/apache/storm/security/auth/ThriftConnectionType.java
b/storm-client/src/jvm/org/apache/storm/security/auth/ThriftConnectionType.java
index 0815194cc..ce7754ced 100644
---
a/storm-client/src/jvm/org/apache/storm/security/auth/ThriftConnectionType.java
+++
b/storm-client/src/jvm/org/apache/storm/security/auth/ThriftConnectionType.java
@@ -12,6 +12,7 @@
package org.apache.storm.security.auth;
+import java.util.Locale;
import java.util.Map;
import org.apache.storm.Config;
import org.apache.storm.generated.WorkerTokenServiceType;
@@ -135,6 +136,14 @@ public enum ThriftConnectionType {
return isFake;
}
+ /**
+ * Prefix for the names of the threads that handle requests of this server
type,
+ * e.g. {@code "nimbus-handler-"}.
+ */
+ public String getHandlerThreadNamePrefix() {
+ return name().toLowerCase(Locale.ROOT) + "-handler-";
+ }
+
public String getTransportPlugin(Map<String, Object> conf) {
String ret = (String) conf.get(transConf);
if (ret == null) {
diff --git
a/storm-client/src/jvm/org/apache/storm/security/auth/sasl/SaslTransportPlugin.java
b/storm-client/src/jvm/org/apache/storm/security/auth/sasl/SaslTransportPlugin.java
index 8f1ad10ba..35ad4aa4a 100644
---
a/storm-client/src/jvm/org/apache/storm/security/auth/sasl/SaslTransportPlugin.java
+++
b/storm-client/src/jvm/org/apache/storm/security/auth/sasl/SaslTransportPlugin.java
@@ -41,6 +41,7 @@ import org.apache.storm.thrift.transport.TTransport;
import org.apache.storm.thrift.transport.TTransportException;
import org.apache.storm.thrift.transport.TTransportFactory;
import org.apache.storm.utils.ExtendedThreadPoolExecutor;
+import org.apache.storm.utils.StormThreadFactory;
/**
* Base class for SASL authentication plugin.
@@ -87,7 +88,8 @@ public abstract class SaslTransportPlugin implements
ITransportPlugin, Closeable
}
ThreadPoolExecutor executorService = new
ExtendedThreadPoolExecutor(numWorkerThreads,
numWorkerThreads,
- 60, TimeUnit.SECONDS, workQueue);
+ 60, TimeUnit.SECONDS, workQueue,
+ StormThreadFactory.create(conf,
type.getHandlerThreadNamePrefix()));
serverArgs.executorService(executorService);
return new TThreadPoolServer(serverArgs);
}
diff --git
a/storm-client/src/jvm/org/apache/storm/security/auth/tls/TlsTransportPlugin.java
b/storm-client/src/jvm/org/apache/storm/security/auth/tls/TlsTransportPlugin.java
index 553141520..337633ce8 100644
---
a/storm-client/src/jvm/org/apache/storm/security/auth/tls/TlsTransportPlugin.java
+++
b/storm-client/src/jvm/org/apache/storm/security/auth/tls/TlsTransportPlugin.java
@@ -42,6 +42,7 @@ import org.apache.storm.thrift.transport.TSocket;
import org.apache.storm.thrift.transport.TTransport;
import org.apache.storm.thrift.transport.TTransportException;
import org.apache.storm.utils.ExtendedThreadPoolExecutor;
+import org.apache.storm.utils.StormThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -111,7 +112,8 @@ public class TlsTransportPlugin implements ITransportPlugin
{
}
ThreadPoolExecutor executorService = new
ExtendedThreadPoolExecutor(numWorkerThreads,
numWorkerThreads,
- 60, TimeUnit.SECONDS, workQueue);
+ 60, TimeUnit.SECONDS, workQueue,
+ StormThreadFactory.create(conf,
type.getHandlerThreadNamePrefix()));
serverArgs.executorService(executorService);
tThreadPoolServer = new TThreadPoolServer(serverArgs);
return tThreadPoolServer;
diff --git
a/storm-client/src/jvm/org/apache/storm/utils/StormThreadFactory.java
b/storm-client/src/jvm/org/apache/storm/utils/StormThreadFactory.java
new file mode 100644
index 000000000..e06df335b
--- /dev/null
+++ b/storm-client/src/jvm/org/apache/storm/utils/StormThreadFactory.java
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.storm.utils;
+
+import java.util.Map;
+import java.util.concurrent.ThreadFactory;
+import org.apache.storm.Config;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Builds the {@link ThreadFactory} used by the Thrift server handler pools.
+ *
+ * <p>When {@link Config#STORM_VIRTUAL_THREADS_ENABLED} is true the factory
creates virtual threads.
+ * Otherwise it creates non-daemon platform threads with normal priority,
matching
+ * {@code Executors.defaultThreadFactory()}. Note that virtual threads are
always daemon threads, so
+ * with the flag on none of these pools keeps the JVM alive. Storm daemons
exit through explicit
+ * {@code System.exit} / {@code Runtime.halt} calls and their shutdown hooks
do not join these
+ * pools, so this makes no difference to them, but it does to an embedding
application that
+ * relies on non-daemon threads to keep running. In both modes threads are
named
+ * {@code <namePrefix><n>} with {@code n} starting at 0; the prefix is used
verbatim, so callers
+ * include their own separator.
+ */
+public final class StormThreadFactory {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(StormThreadFactory.class);
+
+ private StormThreadFactory() {
+ }
+
+ /**
+ * Create a thread factory for a pool.
+ *
+ * @param conf cluster or topology configuration; a raw {@code Map}
is accepted for
+ * callers that hold one
+ * @param namePrefix prefix for the thread names, used verbatim, e.g.
{@code "nimbus-handler-"}
+ * @return a factory producing virtual or platform threads depending on
the configuration
+ */
+ public static ThreadFactory create(Map<?, ?> conf, String namePrefix) {
+ if (isVirtualEnabled(conf)) {
+ return Thread.ofVirtual().name(namePrefix, 0).factory();
+ }
+ return Thread.ofPlatform()
+ .name(namePrefix, 0)
+ .daemon(false)
+ .priority(Thread.NORM_PRIORITY)
+ .factory();
+ }
+
+ /**
+ * Whether virtual threads are enabled for the Thrift server handler pools.
+ *
+ * <p>Boolean values are honoured. String values, which only reach here
through
+ * {@code storm.options} since daemon and topology confs are validated as
boolean, are accepted
+ * when they equal {@code "true"} or {@code "false"} ignoring case;
anything else, like any
+ * other type, is logged and treated as disabled. This method never throws.
+ */
+ public static boolean isVirtualEnabled(Map<?, ?> conf) {
+ if (conf == null) {
+ return false;
+ }
+ Object value = conf.get(Config.STORM_VIRTUAL_THREADS_ENABLED);
+ if (value == null) {
+ return false;
+ }
+ if (value instanceof Boolean) {
+ return (Boolean) value;
+ }
+ if (value instanceof String) {
+ String str = ((String) value).trim();
+ if (str.equalsIgnoreCase("true")) {
+ return true;
+ }
+ if (str.equalsIgnoreCase("false")) {
+ return false;
+ }
+ LOG.warn("Ignoring value '{}' for {}; expected true or false",
value,
+ Config.STORM_VIRTUAL_THREADS_ENABLED);
+ return false;
+ }
+ LOG.warn("Ignoring {} value of type {} for {}; expected a boolean",
value,
+ value.getClass().getName(), Config.STORM_VIRTUAL_THREADS_ENABLED);
+ return false;
+ }
+}
diff --git
a/storm-client/test/jvm/org/apache/storm/security/auth/TlsTransportPluginTest.java
b/storm-client/test/jvm/org/apache/storm/security/auth/TlsTransportPluginTest.java
index 9fa7ba5cb..1f7640ff0 100644
---
a/storm-client/test/jvm/org/apache/storm/security/auth/TlsTransportPluginTest.java
+++
b/storm-client/test/jvm/org/apache/storm/security/auth/TlsTransportPluginTest.java
@@ -17,6 +17,8 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import java.io.IOException;
@@ -28,6 +30,7 @@ import java.security.KeyStore;
import java.security.SecureRandom;
import java.util.HashMap;
import java.util.Map;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLServerSocket;
@@ -37,7 +40,10 @@ import org.apache.storm.Config;
import org.apache.storm.generated.Nimbus;
import org.apache.storm.security.auth.tls.ReloadableTsslTransportFactory;
import org.apache.storm.security.auth.tls.TlsTransportPlugin;
+import org.apache.storm.thrift.protocol.TBinaryProtocol;
import org.apache.storm.thrift.transport.TServerSocket;
+import org.apache.storm.thrift.transport.TSocket;
+import org.apache.storm.utils.Utils;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -150,4 +156,62 @@ class TlsTransportPluginTest {
serverTransport.close();
}
}
-}
\ No newline at end of file
+
+ /**
+ * Round trip through a real {@link ThriftServer} over TLS, asserting from
inside the handler that it runs on a
+ * virtual thread when {@code storm.virtual.threads.enabled} is set. The
TLS transport uses
+ * {@code TThreadPoolServer}, so this covers the executor path that the
simple transport does not. The client is
+ * a plain Thrift client over a trust-only SSL socket because the test
certificate has no subject alternative
+ * name, which the hostname verification in {@code ThriftClient} requires.
+ */
+ @Test
+ void testHandlerRunsOnVirtualThreadWhenEnabled() throws Exception {
+ AtomicReference<Boolean> handlerOnVirtualThread = new
AtomicReference<>();
+ doAnswer(invocation -> {
+ handlerOnVirtualThread.set(Thread.currentThread().isVirtual());
+ return null;
+ }).when(handler).activate(anyString());
+
+ Map<String, Object> serverConf = new
HashMap<>(Utils.readDefaultConfig());
+ serverConf.put(Config.NIMBUS_THRIFT_TLS_TRANSPORT_PLUGIN,
TlsTransportPlugin.class.getName());
+ serverConf.put(Config.NIMBUS_THRIFT_TLS_PORT, 0);
+ serverConf.put(Config.STORM_THRIFT_TLS_SOCKET_TIMEOUT_MS, 5000);
+ serverConf.put(Config.NIMBUS_THRIFT_TLS_SERVER_KEYSTORE_PATH,
testDataPath + "testKeyStore.jks");
+ serverConf.put(Config.NIMBUS_THRIFT_TLS_SERVER_KEYSTORE_PASSWORD,
"testpass");
+ serverConf.put(Config.NIMBUS_THRIFT_TLS_SERVER_TRUSTSTORE_PATH,
testDataPath + "testTrustStore.jks");
+ serverConf.put(Config.NIMBUS_THRIFT_TLS_SERVER_TRUSTSTORE_PASSWORD,
"testpass");
+ serverConf.put(Config.NIMBUS_THRIFT_TLS_CLIENT_AUTH_REQUIRED, false);
+ serverConf.put(Config.NIMBUS_THRIFT_TLS_THREADS, 2);
+ serverConf.put(Config.STORM_VIRTUAL_THREADS_ENABLED, true);
+
+ ThriftServer server = new ThriftServer(serverConf, new
Nimbus.Processor<>(handler), ThriftConnectionType.NIMBUS_TLS);
+ Thread serveThread = new Thread(server::serve, "tls-test-serve");
+ serveThread.setDaemon(true);
+ serveThread.start();
+ try {
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(30);
+ while (!server.isServing()) {
+ assertTrue(System.nanoTime() < deadline, "TLS ThriftServer did
not start serving");
+ Thread.sleep(100);
+ }
+ KeyStore ts = KeyStore.getInstance("JKS");
+ try (InputStream in = Files.newInputStream(Paths.get(testDataPath
+ "testTrustStore.jks"))) {
+ ts.load(in, "testpass".toCharArray());
+ }
+ TrustManagerFactory tmf =
TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
+ tmf.init(ts);
+ SSLContext sslContext = SSLContext.getInstance("TLS");
+ sslContext.init(null, tmf.getTrustManagers(), new SecureRandom());
+ try (SSLSocket sslSocket = (SSLSocket)
sslContext.getSocketFactory()
+ .createSocket(InetAddress.getLocalHost(),
server.getPort())) {
+ sslSocket.setSoTimeout(5000);
+ sslSocket.startHandshake();
+ TSocket transport = new TSocket(sslSocket);
+ new Nimbus.Client(new
TBinaryProtocol(transport)).activate("virtual_thread_tls_test_topology");
+ }
+ assertEquals(Boolean.TRUE, handlerOnVirtualThread.get());
+ } finally {
+ server.stop();
+ }
+ }
+}
diff --git
a/storm-client/test/jvm/org/apache/storm/utils/StormThreadFactoryTest.java
b/storm-client/test/jvm/org/apache/storm/utils/StormThreadFactoryTest.java
new file mode 100644
index 000000000..d818d0b4b
--- /dev/null
+++ b/storm-client/test/jvm/org/apache/storm/utils/StormThreadFactoryTest.java
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.storm.utils;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ThreadFactory;
+
+import org.apache.storm.Config;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class StormThreadFactoryTest {
+
+ @Test
+ public void flagOffCreatesPlatformThreadsWithPrefix() throws Exception {
+ Map<String, Object> conf = new HashMap<>();
+ conf.put(Config.STORM_VIRTUAL_THREADS_ENABLED, false);
+ ThreadFactory factory = StormThreadFactory.create(conf, "test-pool-");
+
+ Thread t = factory.newThread(() -> { });
+ assertFalse(t.isVirtual());
+ assertFalse(t.isDaemon());
+ assertEquals(Thread.NORM_PRIORITY, t.getPriority());
+ assertEquals("test-pool-0", t.getName());
+ assertEquals("test-pool-1", factory.newThread(() -> { }).getName());
+ }
+
+ @Test
+ public void missingFlagBehavesAsOff() {
+ ThreadFactory factory = StormThreadFactory.create(new HashMap<>(),
"test-pool-");
+ assertFalse(factory.newThread(() -> { }).isVirtual());
+ }
+
+ @Test
+ public void flagOnCreatesVirtualThreadsWithPrefix() throws Exception {
+ Map<String, Object> conf = new HashMap<>();
+ conf.put(Config.STORM_VIRTUAL_THREADS_ENABLED, true);
+ ThreadFactory factory = StormThreadFactory.create(conf, "test-pool-");
+
+ Thread t = factory.newThread(() -> { });
+ assertTrue(t.isVirtual());
+ assertEquals("test-pool-0", t.getName());
+ assertEquals("test-pool-1", factory.newThread(() -> { }).getName());
+ }
+
+ @Test
+ public void flagOnAcceptsStringValue() {
+ Map<String, Object> conf = new HashMap<>();
+ conf.put(Config.STORM_VIRTUAL_THREADS_ENABLED, "true");
+ assertTrue(StormThreadFactory.create(conf, "test-pool-").newThread(()
-> { }).isVirtual());
+ }
+
+ @Test
+ public void unrecognizedStringValueBehavesAsOff() {
+ for (String value : new String[]{"yes", "1", "on", ""}) {
+ Map<String, Object> conf = new HashMap<>();
+ conf.put(Config.STORM_VIRTUAL_THREADS_ENABLED, value);
+ assertFalse(StormThreadFactory.isVirtualEnabled(conf), "value '" +
value + "'");
+ }
+ }
+
+ @Test
+ public void platformThreadForcesNormPriorityRegardlessOfCallerPriority()
throws Exception {
+ Map<String, Object> conf = new HashMap<>();
+ conf.put(Config.STORM_VIRTUAL_THREADS_ENABLED, false);
+ ThreadFactory factory = StormThreadFactory.create(conf, "test-pool-");
+
+ Thread callerThread = Thread.currentThread();
+ int originalPriority = callerThread.getPriority();
+ try {
+ callerThread.setPriority(Thread.MAX_PRIORITY);
+ Thread t = factory.newThread(() -> { });
+ assertEquals(Thread.NORM_PRIORITY, t.getPriority());
+ } finally {
+ callerThread.setPriority(originalPriority);
+ }
+ }
+
+ @Test
+ public void unexpectedValueTypeBehavesAsOff() {
+ Map<String, Object> conf = new HashMap<>();
+ conf.put(Config.STORM_VIRTUAL_THREADS_ENABLED, Integer.valueOf(1));
+ ThreadFactory factory = StormThreadFactory.create(conf, "test-pool-");
+ assertFalse(factory.newThread(() -> { }).isVirtual());
+ }
+}
diff --git
a/storm-server/src/test/java/org/apache/storm/security/auth/AuthTest.java
b/storm-server/src/test/java/org/apache/storm/security/auth/AuthTest.java
index cc2e4acae..13fe0d196 100644
--- a/storm-server/src/test/java/org/apache/storm/security/auth/AuthTest.java
+++ b/storm-server/src/test/java/org/apache/storm/security/auth/AuthTest.java
@@ -22,6 +22,7 @@ import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Stream;
import javax.security.auth.Subject;
import org.apache.storm.Config;
import org.apache.storm.Testing;
@@ -45,6 +46,9 @@ import org.apache.storm.utils.NimbusClient;
import org.apache.storm.utils.Time;
import org.apache.storm.utils.Utils;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
import org.junit.jupiter.api.condition.DisabledOnOs;
import org.junit.jupiter.api.condition.OS;
import org.slf4j.Logger;
@@ -647,4 +651,60 @@ public class AuthTest {
public interface MyBiConsumer<T, U> {
void accept(T t, U u) throws Exception;
}
-}
\ No newline at end of file
+
+ /**
+ * Each case runs a real ThriftServer with the given transport and extra
conf, calls it through NimbusClient
+ * and asserts from inside the handler whether it ran on a virtual thread.
The principal check only applies to
+ * the authenticated transport, where it proves ReqContext is still
populated on a virtual handler thread.
+ */
+ static Stream<Arguments> handlerThreadCases() {
+ Map<String, Object> virtualOn = new HashMap<>();
+ virtualOn.put(Config.STORM_VIRTUAL_THREADS_ENABLED, true);
+
+ // defaults.yaml ships nimbus.queue.size, so unset it explicitly to
exercise the path where
+ // THsHaServer would otherwise build its own pool (the DRPC
invocations case).
+ Map<String, Object> virtualOnNoQueue = new HashMap<>(virtualOn);
+ virtualOnNoQueue.put(Config.NIMBUS_QUEUE_SIZE, null);
+
+ Map<String, Object> virtualOnSmallQueue = new HashMap<>(virtualOn);
+ virtualOnSmallQueue.put(Config.NIMBUS_QUEUE_SIZE, 8);
+
+ return Stream.of(
+ Arguments.of("simple transport, no queue size, flag on", null,
SimpleTransportPlugin.class,
+ virtualOnNoQueue, true, null),
+ Arguments.of("simple transport, queue size, flag on", null,
SimpleTransportPlugin.class,
+ virtualOnSmallQueue, true, null),
+ Arguments.of("simple transport, flag off by default", null,
SimpleTransportPlugin.class,
+ null, false, null),
+ Arguments.of("digest transport, flag on", DIGEST_JAAS_CONF,
DigestSaslTransportPlugin.class,
+ virtualOn, true, "bob"));
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("handlerThreadCases")
+ public void handlerRunsOnExpectedThreadType(String name, String loginCfg,
Class<? extends ITransportPlugin> transport,
+ Map<String, Object> extra,
boolean expectVirtual,
+ String expectedPrincipal)
throws Exception {
+ Nimbus.Iface impl = mock(Nimbus.Iface.class);
+ final AtomicReference<Boolean> handlerOnVirtualThread = new
AtomicReference<>();
+ final AtomicReference<ReqContext> user = new AtomicReference<>();
+ doAnswer((invocation) -> {
+ handlerOnVirtualThread.set(Thread.currentThread().isVirtual());
+ user.set(new ReqContext(ReqContext.context()));
+ return null;
+ }).when(impl).activate(anyString());
+
+ withServer(loginCfg, transport, impl, null, extra,
+ (ThriftServer server, Map<String, Object> conf) -> {
+ try (NimbusClient client =
NimbusClient.Builder.withConf(conf).withTimeout(NIMBUS_TIMEOUT)
+ .buildWithNimbusHostPort("localhost",
server.getPort())) {
+
client.getClient().activate("handler_thread_test_topology");
+ }
+ assertEquals(expectVirtual, handlerOnVirtualThread.get());
+ if (expectedPrincipal != null) {
+ assertNotNull(user.get());
+ assertEquals(expectedPrincipal,
user.get().principal().getName());
+ }
+ });
+ }
+}
