diff --git a/activemq-client/src/main/java/org/apache/activemq/ActiveMQConnectionFactory.java b/activemq-client/src/main/java/org/apache/activemq/ActiveMQConnectionFactory.java index 93cfb20d309..4eb75fc798a 100644 --- a/activemq-client/src/main/java/org/apache/activemq/ActiveMQConnectionFactory.java +++ b/activemq-client/src/main/java/org/apache/activemq/ActiveMQConnectionFactory.java @@ -43,6 +43,7 @@ import org.apache.activemq.transport.Transport; import org.apache.activemq.transport.TransportFactory; import org.apache.activemq.transport.TransportListener; +import org.apache.activemq.transport.failover.FailoverTransport; import org.apache.activemq.util.*; import org.apache.activemq.util.URISupport.CompositeData; import org.slf4j.Logger; @@ -392,6 +393,17 @@ protected ActiveMQConnection createActiveMQConnection(String userName, String pa configureConnection(connection); + // The failover reconnect task is created on first use, so the connection level + // virtual thread preference has to reach the transport before it starts. A value + // set on the failover URI itself takes precedence and is left alone, and a factory + // left at false says nothing, so the system property default still applies. + if (isUseVirtualThreadTaskRunner()) { + FailoverTransport failover = transport.narrow(FailoverTransport.class); + if (failover != null && failover.getUseVirtualThreadTaskRunner() == null) { + failover.setUseVirtualThreadTaskRunner(true); + } + } + transport.start(); if (clientID != null) { @@ -906,6 +918,8 @@ public void populateProperties(Properties props) { props.setProperty("transactedIndividualAck", Boolean.toString(isTransactedIndividualAck())); props.setProperty("nonBlockingRedelivery", Boolean.toString(isNonBlockingRedelivery())); props.setProperty("maxThreadPoolSize", Integer.toString(getMaxThreadPoolSize())); + props.setProperty("useDedicatedTaskRunner", Boolean.toString(isUseDedicatedTaskRunner())); + props.setProperty("useVirtualThreadTaskRunner", Boolean.toString(isUseVirtualThreadTaskRunner())); props.setProperty("nestedMapAndListEnabled", Boolean.toString(isNestedMapAndListEnabled())); props.setProperty("consumerFailoverRedeliveryWaitPeriod", Long.toString(getConsumerFailoverRedeliveryWaitPeriod())); props.setProperty("rmIdFromConnectionId", Boolean.toString(isRmIdFromConnectionId())); diff --git a/activemq-client/src/main/java/org/apache/activemq/thread/TaskRunnerFactory.java b/activemq-client/src/main/java/org/apache/activemq/thread/TaskRunnerFactory.java index 28f834efcbe..1904fd566ae 100644 --- a/activemq-client/src/main/java/org/apache/activemq/thread/TaskRunnerFactory.java +++ b/activemq-client/src/main/java/org/apache/activemq/thread/TaskRunnerFactory.java @@ -58,6 +58,9 @@ public class TaskRunnerFactory implements Executor { private final AtomicBoolean initDone = new AtomicBoolean(false); private int maxThreadPoolSize = getDefaultMaximumPoolSize(); private RejectedExecutionHandler rejectedTaskHandler = null; + /** system property that turns on virtual thread task runners for every factory that is not told otherwise */ + public static final String USE_VIRTUAL_THREAD_TASK_RUNNER_PROPERTY = "org.apache.activemq.UseVirtualThreadTaskRunner"; + private ClassLoader threadClassLoader; // [AMQ-9394] Virtual Thread support @@ -106,7 +109,7 @@ public void init() { synchronized(this) { //need to recheck if initDone is true under the lock if (!initDone.get()) { - if (virtualThreadTaskRunner || "true".equalsIgnoreCase(System.getProperty("org.apache.activemq.UseVirtualThreadTaskRunner")) ) { + if (virtualThreadTaskRunner || isVirtualThreadTaskRunnerDefault()) { executorRef.compareAndSet(null, createVirtualThreadExecutor()); } else if (dedicatedTaskRunner || "true".equalsIgnoreCase(System.getProperty("org.apache.activemq.UseDedicatedTaskRunner"))) { executorRef.set(null); @@ -239,11 +242,14 @@ public void uncaughtException(final Thread t, final Throwable e) { return rc; } + // The JDK 21 classes ship in this artifact under META-INF/versions/21, so they are + // loaded through this class's own loader. threadClassLoader is the context loader + // for spawned threads and is unset on the client, where it would mean the bootstrap loader. protected ExecutorService createVirtualThreadExecutor() { assertJDK21VirtualThreadSupport(); try { - Class virtualThreadExecutorClass = Class.forName("org.apache.activemq.thread.VirtualThreadExecutor", false, threadClassLoader); + Class virtualThreadExecutorClass = Class.forName("org.apache.activemq.thread.VirtualThreadExecutor", false, TaskRunnerFactory.class.getClassLoader()); Method method = virtualThreadExecutorClass.getMethod("createVirtualThreadExecutorService", String.class, AtomicLong.class, Logger.class); Object result = method.invoke(null, name, id, LOG); if(!ExecutorService.class.isAssignableFrom(result.getClass())) { @@ -261,7 +267,7 @@ protected TaskRunner createVirtualThreadTaskRunner(Executor executor, Task task, assertJDK21VirtualThreadSupport(); try { - Class virtualThreadTaskRunnerClass = Class.forName("org.apache.activemq.thread.VirtualThreadTaskRunner", false, threadClassLoader); + Class virtualThreadTaskRunnerClass = Class.forName("org.apache.activemq.thread.VirtualThreadTaskRunner", false, TaskRunnerFactory.class.getClassLoader()); Constructor constructor = virtualThreadTaskRunnerClass.getConstructor(Executor.class, Task.class, Integer.TYPE); Object result = constructor.newInstance(executor, task, maxIterations); if(!TaskRunner.class.isAssignableFrom(result.getClass())) { @@ -274,6 +280,11 @@ protected TaskRunner createVirtualThreadTaskRunner(Executor executor, Task task, } } + /** the process wide default from {@link #USE_VIRTUAL_THREAD_TASK_RUNNER_PROPERTY} */ + public static boolean isVirtualThreadTaskRunnerDefault() { + return "true".equalsIgnoreCase(System.getProperty(USE_VIRTUAL_THREAD_TASK_RUNNER_PROPERTY)); + } + private void assertJDK21VirtualThreadSupport() { if(!(Runtime.version().feature() >= 21)) { LOG.error("Virtual Thread support requires JDK 21 or higher"); diff --git a/activemq-client/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java b/activemq-client/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java index 5be8cf3b5ce..d21c5c56c59 100644 --- a/activemq-client/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java +++ b/activemq-client/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java @@ -92,8 +92,11 @@ public class FailoverTransport implements CompositeTransport { private URI connectedTransportURI; private URI failedConnectTransportURI; private final AtomicReference connectedTransport = new AtomicReference(); - private final TaskRunnerFactory reconnectTaskFactory; - private final TaskRunner reconnectTask; + private final Object reconnectTaskMutex = new Object(); + private TaskRunnerFactory reconnectTaskFactory; + private volatile TaskRunner reconnectTask; + // null until set by a failover URI option; the connection factory fills it in only then + private Boolean useVirtualThreadTaskRunner; private volatile boolean started; private long initialReconnectDelay = DEFAULT_INITIAL_RECONNECT_DELAY; private long maxReconnectDelay = 1000 * 30; @@ -140,10 +143,23 @@ public FailoverTransport() { public FailoverTransport(SslContext sslContext) { this.sslContext = sslContext; stateTracker.setTrackTransactions(true); - // Setup a task that is used to reconnect the a connection async. - reconnectTaskFactory = new TaskRunnerFactory(); - reconnectTaskFactory.init(); - reconnectTask = reconnectTaskFactory.createTaskRunner(new Task() { + } + + /** + * The reconnect task is built on first use rather than in the constructor so + * that URI options and the owning connection factory can decide whether it + * runs on a virtual thread before any thread is created. + */ + private TaskRunner getReconnectTask() { + TaskRunner task = reconnectTask; + if (task == null) { + synchronized (reconnectTaskMutex) { + task = reconnectTask; + if (task == null) { + reconnectTaskFactory = new TaskRunnerFactory("ActiveMQ Task", Thread.NORM_PRIORITY, true, 1000, + false, isUseVirtualThreadTaskRunner()); + reconnectTaskFactory.init(); + task = reconnectTaskFactory.createTaskRunner(new Task() { @Override public boolean iterate() { boolean result = false; @@ -162,10 +178,7 @@ public boolean iterate() { if (priorityBackup && !connectedToPriority) { try { doDelay(); - if (reconnectTask == null) { - return true; - } - reconnectTask.wakeup(); + getReconnectTask().wakeup(); } catch (InterruptedException e) { LOG.debug("Reconnect task has been interrupted.", e); } @@ -174,10 +187,7 @@ public boolean iterate() { // build backups on the next iteration buildBackup = true; try { - if (reconnectTask == null) { - return true; - } - reconnectTask.wakeup(); + getReconnectTask().wakeup(); } catch (InterruptedException e) { LOG.debug("Reconnect task has been interrupted.", e); } @@ -186,6 +196,11 @@ public boolean iterate() { } }, "ActiveMQ Failover Worker: " + System.identityHashCode(this)); + reconnectTask = task; + } + } + } + return task; } private void processCommand(Object incoming) { @@ -298,7 +313,7 @@ public final void handleTransportFailure(Transport failed, IOException e) throws transportListener.transportInterupted(); } - reconnectTask.wakeup(); + getReconnectTask().wakeup(); } else if (!isDisposed()) { propagateFailureToExceptionListener(e); } @@ -408,8 +423,20 @@ public void stop() throws Exception { sleepMutex.notifyAll(); } } finally { - reconnectTask.shutdown(); - reconnectTaskFactory.shutdownNow(); + // Take the references under the mutex but shut down outside it: shutdown() + // waits for a running iteration, and that iteration may call getReconnectTask() + TaskRunner taskToStop; + TaskRunnerFactory factoryToStop; + synchronized (reconnectTaskMutex) { + taskToStop = reconnectTask; + factoryToStop = reconnectTaskFactory; + } + if (taskToStop != null) { + taskToStop.shutdown(); + } + if (factoryToStop != null) { + factoryToStop.shutdownNow(); + } } synchronized(backupMutex) { @@ -796,7 +823,7 @@ public void reconnect(boolean rebalance) { } LOG.debug("Waking up reconnect task"); try { - reconnectTask.wakeup(); + getReconnectTask().wakeup(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } @@ -1449,6 +1476,31 @@ private URI addExtraQueryOptions(URI uri) { return uri; } + /** + * The effective setting: an explicit value from the URI or the connection + * factory, otherwise the process wide default from the + * {@code org.apache.activemq.UseVirtualThreadTaskRunner} system property. + */ + public boolean isUseVirtualThreadTaskRunner() { + return useVirtualThreadTaskRunner != null ? useVirtualThreadTaskRunner : TaskRunnerFactory.isVirtualThreadTaskRunnerDefault(); + } + + /** the raw setting, null when neither a URI option nor the connection factory has set it */ + public Boolean getUseVirtualThreadTaskRunner() { + return useVirtualThreadTaskRunner; + } + + /** + * Run the reconnect task on a virtual thread (JDK 21+). A value given as a + * failover URI option is final; the owning connection factory turns this on + * when its own useVirtualThreadTaskRunner is set and the URI said nothing; + * left unset, the system property default applies. Takes effect only before + * the transport starts. + */ + public void setUseVirtualThreadTaskRunner(Boolean useVirtualThreadTaskRunner) { + this.useVirtualThreadTaskRunner = useVirtualThreadTaskRunner; + } + public void setNestedExtraQueryOptions(String nestedExtraQueryOptions) { this.nestedExtraQueryOptions = nestedExtraQueryOptions; } diff --git a/activemq-client/src/main/java21/org/apache/activemq/thread/VirtualThreadTaskRunner.java b/activemq-client/src/main/java21/org/apache/activemq/thread/VirtualThreadTaskRunner.java index 0b58657e7ea..0607c8760d7 100644 --- a/activemq-client/src/main/java21/org/apache/activemq/thread/VirtualThreadTaskRunner.java +++ b/activemq-client/src/main/java21/org/apache/activemq/thread/VirtualThreadTaskRunner.java @@ -109,7 +109,13 @@ public void shutdown(long timeout) throws InterruptedException { // waiting for iterating to finish if (runningThread != Thread.currentThread()) { if (iterating) { - runnableCondition.await(timeout, TimeUnit.MILLISECONDS); + // Same contract as Object.wait(timeout) in the other runners: + // zero means wait until the running iteration signals, not return at once + if (timeout > 0) { + runnableCondition.await(timeout, TimeUnit.MILLISECONDS); + } else { + runnableCondition.await(); + } } } } finally { diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/ActiveMQConnectionFactoryTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/ActiveMQConnectionFactoryTest.java index 42a154739a5..9088feb9297 100644 --- a/activemq-unit-tests/src/test/java/org/apache/activemq/ActiveMQConnectionFactoryTest.java +++ b/activemq-unit-tests/src/test/java/org/apache/activemq/ActiveMQConnectionFactoryTest.java @@ -215,6 +215,27 @@ public void testFactorySerializable() throws Exception { assertEquals(cf.getClientID(), clientID); } + public void testTaskRunnerSettingsSurviveSerialization() throws Exception { + // Serialization and JNDI both go through populateProperties/buildFromProperties, + // so a setting missing from populateProperties is silently reset on the way back + var cf = new ActiveMQConnectionFactory(); + cf.setUseDedicatedTaskRunner(true); + cf.setUseVirtualThreadTaskRunner(true); + cf.setMaxThreadPoolSize(7); + + var bytesOut = new ByteArrayOutputStream(); + try (var objectsOut = new ObjectOutputStream(bytesOut)) { + objectsOut.writeObject(cf); + } + try (var objectsIn = new ObjectInputStream(new ByteArrayInputStream(bytesOut.toByteArray()))) { + cf = (ActiveMQConnectionFactory) objectsIn.readObject(); + } + + assertTrue("useDedicatedTaskRunner lost", cf.isUseDedicatedTaskRunner()); + assertTrue("useVirtualThreadTaskRunner lost", cf.isUseVirtualThreadTaskRunner()); + assertEquals(7, cf.getMaxThreadPoolSize()); + } + public void testSetExceptionListener() throws Exception { ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false"); connection = (ActiveMQConnection)cf.createConnection(); diff --git a/activemq-unit-tests/src/test/java21/org/apache/activemq/VirtualThreadConfigurationMatrixTest.java b/activemq-unit-tests/src/test/java21/org/apache/activemq/VirtualThreadConfigurationMatrixTest.java new file mode 100644 index 00000000000..d5a0aaf4b48 --- /dev/null +++ b/activemq-unit-tests/src/test/java21/org/apache/activemq/VirtualThreadConfigurationMatrixTest.java @@ -0,0 +1,189 @@ +/** + * 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.activemq; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +import java.lang.reflect.Field; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeUnit; + +import org.apache.activemq.broker.BrokerService; +import org.apache.activemq.test.annotations.ParallelTest; +import org.apache.activemq.thread.TaskRunnerFactory; +import org.apache.activemq.transport.failover.FailoverTransport; +import org.junit.After; +import org.junit.AfterClass; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; + +/** + * Every way of asking for virtual thread task runners, combined, with the + * resulting field on each object that acts on the answer: + * + */ +@Category(ParallelTest.class) +@RunWith(Parameterized.class) +public class VirtualThreadConfigurationMatrixTest { + + /** how the connection factory flag is set */ + enum FactorySetting { + UNSET, SETTER_TRUE, SETTER_FALSE, URI_TRUE, URI_FALSE, URI_TRUE_THEN_SETTER_FALSE + } + + private static final String PROPERTY = TaskRunnerFactory.USE_VIRTUAL_THREAD_TASK_RUNNER_PROPERTY; + private static BrokerService broker; + + @Parameterized.Parameters(name = "property={0} factory={1} transportUri={2}") + public static Collection matrix() { + List rows = new ArrayList<>(); + for (boolean property : new boolean[] {false, true}) { + for (FactorySetting factory : FactorySetting.values()) { + for (Boolean transportUri : new Boolean[] {null, Boolean.TRUE, Boolean.FALSE}) { + rows.add(new Object[] {property, factory, transportUri}); + } + } + } + return rows; + } + + @Parameterized.Parameter(0) + public boolean property; + + @Parameterized.Parameter(1) + public FactorySetting factory; + + @Parameterized.Parameter(2) + public Boolean transportUri; + + private String previousProperty; + + @BeforeClass + public static void startBroker() throws Exception { + broker = new BrokerService(); + broker.setPersistent(false); + broker.setUseJmx(false); + broker.start(); + broker.waitUntilStarted(); + } + + @AfterClass + public static void stopBroker() throws Exception { + broker.stop(); + broker.waitUntilStopped(); + } + + @Before + public void setProperty() { + previousProperty = System.getProperty(PROPERTY); + if (property) { + System.setProperty(PROPERTY, "true"); + } else { + System.clearProperty(PROPERTY); + } + } + + @After + public void restoreProperty() { + if (previousProperty == null) { + System.clearProperty(PROPERTY); + } else { + System.setProperty(PROPERTY, previousProperty); + } + } + + @Test(timeout = 30000) + public void fieldsOnEveryObjectFollowThePrecedence() throws Exception { + var url = new StringBuilder("failover:(vm://localhost?create=false)?initialReconnectDelay=10&maxReconnectDelay=100"); + if (factory == FactorySetting.URI_TRUE || factory == FactorySetting.URI_TRUE_THEN_SETTER_FALSE) { + url.append("&jms.useVirtualThreadTaskRunner=true"); + } else if (factory == FactorySetting.URI_FALSE) { + url.append("&jms.useVirtualThreadTaskRunner=false"); + } + if (transportUri != null) { + url.append("&useVirtualThreadTaskRunner=").append(transportUri); + } + + var connectionFactory = new ActiveMQConnectionFactory(url.toString()); + if (factory == FactorySetting.SETTER_TRUE) { + connectionFactory.setUseVirtualThreadTaskRunner(true); + } else if (factory == FactorySetting.SETTER_FALSE || factory == FactorySetting.URI_TRUE_THEN_SETTER_FALSE) { + connectionFactory.setUseVirtualThreadTaskRunner(false); + } + + boolean expectedFactory = factory == FactorySetting.SETTER_TRUE || factory == FactorySetting.URI_TRUE; + assertEquals("connection factory field", expectedFactory, connectionFactory.isUseVirtualThreadTaskRunner()); + + try (var connection = (ActiveMQConnection) connectionFactory.createConnection()) { + connection.start(); + assertEquals("connection field copies the factory", expectedFactory, connection.isUseVirtualThreadTaskRunner()); + + // the transport: a URI value is final, a factory true fills a gap, otherwise the property + var failover = connection.getTransport().narrow(FailoverTransport.class); + Boolean expectedRecorded = transportUri != null ? transportUri : (expectedFactory ? Boolean.TRUE : null); + boolean expectedTransport = transportUri != null ? transportUri : (expectedFactory || property); + assertEquals("value recorded on the transport", expectedRecorded, failover.getUseVirtualThreadTaskRunner()); + assertEquals("effective transport setting", expectedTransport, failover.isUseVirtualThreadTaskRunner()); + + // start() creates the reconnect runner, so its factory carries the transport's answer + var reconnectFactory = (TaskRunnerFactory) field(FailoverTransport.class, "reconnectTaskFactory").get(failover); + assertNotNull("reconnect task factory created on start", reconnectFactory); + assertEquals("reconnect factory flag", expectedTransport, reconnectFactory.isVirtualThreadTaskRunner()); + // TaskRunnerFactory.init() applies the system property on its own, so the executor is + // virtual whenever either the flag or the property says so + assertEquals("reconnect executor", expectedTransport || property, runsOnVirtualThread(reconnectFactory.getExecutor())); + + var sessionFactory = connection.getSessionTaskRunner(); + sessionFactory.init(); + assertEquals("session factory flag", expectedFactory, sessionFactory.isVirtualThreadTaskRunner()); + assertEquals("session executor", expectedFactory || property, runsOnVirtualThread(sessionFactory.getExecutor())); + } + } + + /** + * Asks the executor itself: a task submitted to it reports whether it ran on + * a virtual thread. Thread.isVirtual() is read reflectively because this + * module compiles its tests for an older release. + */ + private static boolean runsOnVirtualThread(ExecutorService executor) throws Exception { + assertNotNull("executor initialised", executor); + return executor.submit(() -> (Boolean) Thread.class.getMethod("isVirtual").invoke(Thread.currentThread())) + .get(10, TimeUnit.SECONDS); + } + + private static Field field(Class type, String name) throws NoSuchFieldException { + Field field = type.getDeclaredField(name); + field.setAccessible(true); + return field; + } +} diff --git a/activemq-unit-tests/src/test/java21/org/apache/activemq/VirtualThreadSessionTaskRunnerTest.java b/activemq-unit-tests/src/test/java21/org/apache/activemq/VirtualThreadSessionTaskRunnerTest.java new file mode 100644 index 00000000000..a4ff9b3fdea --- /dev/null +++ b/activemq-unit-tests/src/test/java21/org/apache/activemq/VirtualThreadSessionTaskRunnerTest.java @@ -0,0 +1,327 @@ +/** + * 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.activemq; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import java.io.IOException; +import java.net.URI; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; + +import jakarta.jms.Session; + +import org.apache.activemq.broker.BrokerService; +import org.apache.activemq.command.ActiveMQQueue; +import org.apache.activemq.test.annotations.ParallelTest; +import org.apache.activemq.thread.TaskRunnerFactory; +import org.apache.activemq.transport.TransportFactory; +import org.apache.activemq.transport.TransportListener; +import org.apache.activemq.transport.failover.FailoverTransport; +import org.apache.activemq.util.Wait; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.junit.experimental.categories.Category; + +/** + * Client side coverage for {@code useVirtualThreadTaskRunner}: message + * listeners run on virtual threads when the flag is set via the broker URL, + * and {@code Connection.stop()} keeps its JMS contract of blocking until an + * in-progress listener returns. + */ +@Category(ParallelTest.class) +public class VirtualThreadSessionTaskRunnerTest { + + private static final ActiveMQQueue QUEUE = new ActiveMQQueue("TEST.VIRTUAL.THREAD"); + + private BrokerService broker; + + @Before + public void setUp() throws Exception { + broker = new BrokerService(); + broker.setPersistent(false); + broker.setUseJmx(false); + broker.start(); + broker.waitUntilStarted(); + } + + @After + public void tearDown() throws Exception { + broker.stop(); + broker.waitUntilStopped(); + } + + @Test(timeout = 30000) + public void testListenerRunsOnVirtualThreadWhenEnabled() throws Exception { + assertTrue(dispatchThreadIsVirtual(true)); + } + + @Test(timeout = 30000) + public void testListenerRunsOnPlatformThreadByDefault() throws Exception { + assertFalse(dispatchThreadIsVirtual(false)); + } + + /** + * JMS requires stop() to block until message listeners in progress have + * completed. The consumer dispatches under its channel mutex and stop() + * takes the same mutex, so this holds however the session task is run; + * the test confirms the contract end to end with virtual thread dispatch. + */ + @Test(timeout = 30000) + public void testStopWaitsForInProgressListener() throws Exception { + var listenerEntered = new CountDownLatch(1); + var releaseListener = new CountDownLatch(1); + var stopReturned = new CountDownLatch(1); + var stopError = new AtomicReference(); + + try (var connection = factory(true).createConnection(); + var session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + var consumer = session.createConsumer(QUEUE); + var producer = session.createProducer(QUEUE)) { + + consumer.setMessageListener(message -> { + listenerEntered.countDown(); + try { + releaseListener.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }); + connection.start(); + producer.send(session.createTextMessage("block")); + assertTrue("listener never invoked", listenerEntered.await(10, TimeUnit.SECONDS)); + + var stopper = new Thread(() -> { + try { + connection.stop(); + } catch (Throwable t) { + stopError.set(t); + } finally { + stopReturned.countDown(); + } + }, "connection-stopper"); + stopper.start(); + + assertFalse("stop() returned while the listener was still running", + stopReturned.await(500, TimeUnit.MILLISECONDS)); + + releaseListener.countDown(); + assertTrue("stop() did not return after the listener completed", + stopReturned.await(10, TimeUnit.SECONDS)); + assertEquals(null, stopError.get()); + } + } + + /** + * TaskRunner.shutdown() must block until a running iterate() returns, the + * same as PooledTaskRunner. A Condition.await(0) returns at once where + * Object.wait(0) waits, so the virtual runner has to special case zero. + */ + @Test(timeout = 30000) + public void testVirtualRunnerShutdownWaitsForRunningIteration() throws Exception { + assertShutdownWaitsForIteration(true); + } + + @Test(timeout = 30000) + public void testPooledRunnerShutdownWaitsForRunningIteration() throws Exception { + assertShutdownWaitsForIteration(false); + } + + private void assertShutdownWaitsForIteration(boolean useVirtualThreadTaskRunner) throws Exception { + var factory = new TaskRunnerFactory("ActiveMQ Test Task", Thread.NORM_PRIORITY, true, 1000, false, + useVirtualThreadTaskRunner); + var iterating = new CountDownLatch(1); + var release = new CountDownLatch(1); + var shutdownReturned = new CountDownLatch(1); + try { + var runner = factory.createTaskRunner(() -> { + iterating.countDown(); + try { + release.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return false; + }, "blocking-task"); + runner.wakeup(); + assertTrue("iterate() never started", iterating.await(10, TimeUnit.SECONDS)); + + var stopper = new Thread(() -> { + try { + runner.shutdown(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + shutdownReturned.countDown(); + } + }, "runner-stopper"); + stopper.start(); + + assertFalse("shutdown() returned while iterate() was still running", + shutdownReturned.await(500, TimeUnit.MILLISECONDS)); + + release.countDown(); + assertTrue("shutdown() did not return after iterate() completed", + shutdownReturned.await(10, TimeUnit.SECONDS)); + } finally { + factory.shutdownNow(); + } + } + + /** + * The failover reconnect task is a separate TaskRunnerFactory. The factory + * level flag has to reach it, and transportResumed() is delivered from the + * reconnect task's thread, which is how the test observes it. + */ + @Test(timeout = 60000) + public void testFailoverReconnectRunsOnVirtualThreadWhenEnabled() throws Exception { + assertTrue(failoverReconnectThreadIsVirtual(true, null)); + } + + @Test(timeout = 60000) + public void testFailoverReconnectRunsOnPlatformThreadByDefault() throws Exception { + assertFalse(failoverReconnectThreadIsVirtual(false, null)); + } + + /** a value on the failover URI is final in both directions; the factory flag only fills a gap */ + @Test(timeout = 60000) + public void testFailoverUriOptionOverridesFactoryFlag() throws Exception { + assertFalse("URI false must win over the factory's true", failoverReconnectThreadIsVirtual(true, false)); + assertTrue("URI true must win over the factory's false", failoverReconnectThreadIsVirtual(false, true)); + } + + /** + * A transport built outside a connection factory has nothing pushing a value, so + * the system property is the default; a URI value still wins over the property. + */ + @Test(timeout = 30000) + public void testUnsetTransportFollowsSystemPropertyDefault() throws Exception { + var previous = System.getProperty(TaskRunnerFactory.USE_VIRTUAL_THREAD_TASK_RUNNER_PROPERTY); + System.setProperty(TaskRunnerFactory.USE_VIRTUAL_THREAD_TASK_RUNNER_PROPERTY, "true"); + try { + var unset = (FailoverTransport) TransportFactory.connect(new URI("failover:(vm://localhost?create=false)")).narrow(FailoverTransport.class); + assertEquals("nothing recorded", null, unset.getUseVirtualThreadTaskRunner()); + assertTrue("property default applies", unset.isUseVirtualThreadTaskRunner()); + + var uriFalse = (FailoverTransport) TransportFactory.connect(new URI("failover:(vm://localhost?create=false)?useVirtualThreadTaskRunner=false")).narrow(FailoverTransport.class); + assertFalse("URI false wins over the property", uriFalse.isUseVirtualThreadTaskRunner()); + } finally { + if (previous == null) { + System.clearProperty(TaskRunnerFactory.USE_VIRTUAL_THREAD_TASK_RUNNER_PROPERTY); + } else { + System.setProperty(TaskRunnerFactory.USE_VIRTUAL_THREAD_TASK_RUNNER_PROPERTY, previous); + } + } + } + + private boolean failoverReconnectThreadIsVirtual(boolean factoryFlag, Boolean transportOption) throws Exception { + var url = "failover:(vm://localhost?create=false)?initialReconnectDelay=10&maxReconnectDelay=100" + + "&jms.useVirtualThreadTaskRunner=" + factoryFlag + + (transportOption == null ? "" : "&useVirtualThreadTaskRunner=" + transportOption); + var expected = transportOption != null ? transportOption : factoryFlag; + // a factory left at false says nothing, so the transport records only true or the URI value + var recorded = transportOption != null ? transportOption : (factoryFlag ? Boolean.TRUE : null); + var resumed = new CountDownLatch(1); + var virtual = new AtomicBoolean(); + + try (var connection = (ActiveMQConnection) new ActiveMQConnectionFactory(url).createConnection()) { + connection.start(); + var failover = connection.getTransport().narrow(FailoverTransport.class); + assertEquals(expected, failover.isUseVirtualThreadTaskRunner()); + assertEquals("what was recorded on the transport", recorded, failover.getUseVirtualThreadTaskRunner()); + + connection.addTransportListener(new TransportListener() { + @Override + public void onCommand(Object command) { + } + + @Override + public void onException(IOException error) { + } + + @Override + public void transportInterupted() { + } + + @Override + public void transportResumed() { + virtual.set(isVirtual(Thread.currentThread())); + resumed.countDown(); + } + }); + + // Bounce the broker so failover has to reconnect + restartBroker(); + assertTrue("failover never reconnected", resumed.await(30, TimeUnit.SECONDS)); + } + return virtual.get(); + } + + private void restartBroker() throws Exception { + broker.stop(); + broker.waitUntilStopped(); + broker = new BrokerService(); + broker.setPersistent(false); + broker.setUseJmx(false); + broker.start(); + broker.waitUntilStarted(); + } + + private boolean dispatchThreadIsVirtual(boolean useVirtualThreadTaskRunner) throws Exception { + var received = new CountDownLatch(1); + var virtual = new AtomicBoolean(); + + try (var connection = factory(useVirtualThreadTaskRunner).createConnection(); + var session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + var consumer = session.createConsumer(QUEUE); + var producer = session.createProducer(QUEUE)) { + + consumer.setMessageListener(message -> { + virtual.set(isVirtual(Thread.currentThread())); + received.countDown(); + }); + connection.start(); + producer.send(session.createTextMessage("probe")); + assertTrue("message not delivered", received.await(10, TimeUnit.SECONDS)); + assertTrue(Wait.waitFor(() -> received.getCount() == 0, 1000, 10)); + } + return virtual.get(); + } + + /** + * Thread.isVirtual() is a JDK 21 API and this module compiles tests for an + * older release, so reach it reflectively. This source root is only added + * to the build on JDK 21 and later. + */ + private static boolean isVirtual(Thread thread) { + try { + return (Boolean) Thread.class.getMethod("isVirtual").invoke(thread); + } catch (ReflectiveOperationException e) { + throw new IllegalStateException("Thread.isVirtual() unavailable; JDK 21+ required", e); + } + } + + private static ActiveMQConnectionFactory factory(boolean useVirtualThreadTaskRunner) { + return new ActiveMQConnectionFactory( + "vm://localhost?create=false&jms.useVirtualThreadTaskRunner=" + useVirtualThreadTaskRunner); + } +}