You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tomee.apache.org by rz...@apache.org on 2026/04/02 13:31:21 UTC
(tomee) branch concurency updated (41dab0e655 -> 40716ae895)
This is an automated email from the ASF dual-hosted git repository.
rzo1 pushed a change to branch concurency
in repository https://gitbox.apache.org/repos/asf/tomee.git
from 41dab0e655 Add CronTrigger tests verifying ZonedTrigger works with existing scheduler
new bdc3fe9395 Fix scheduled async interceptor to call setFuture before ctx.proceed
new a830d73543 Fix scheduled async lifecycle: stop on non-null return, reject invalid JNDI
new 8447b7f17d Use bean-discovery-mode=annotated in Concurrency TCK archive processor
new 020017b41d Add multiple schedules and maxAsync unit tests for scheduled async
new 40716ae895 Fix JNDI lookup for java:module/ and java:app/ scoped scheduled executors
The 5 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../ScheduledAsyncCustomExecutorTest.java | 110 +++++++++
.../cdi/concurrency/AsynchronousInterceptor.java | 66 +++--
...ManagedScheduledExecutorServiceImplFactory.java | 31 ++-
.../AsynchronousScheduledTCKStyleTest.java | 270 +++++++++++++++++++++
.../ConcurrencyTCKArchiveProcessor.java | 56 +++++
.../tck/concurrency/ConcurrencyTCKExtension.java} | 10 +-
...org.jboss.arquillian.core.spi.LoadableExtension | 1 +
7 files changed, 504 insertions(+), 40 deletions(-)
create mode 100644 arquillian/arquillian-tomee-tests/arquillian-tomee-webprofile-tests/src/test/java/org/apache/openejb/arquillian/tests/concurrency/ScheduledAsyncCustomExecutorTest.java
create mode 100644 container/openejb-core/src/test/java/org/apache/openejb/cdi/concurrency/AsynchronousScheduledTCKStyleTest.java
create mode 100644 tck/concurrency-standalone/src/test/java/org/apache/tomee/tck/concurrency/ConcurrencyTCKArchiveProcessor.java
copy tck/{data-standalone/src/test/java/org/apache/tomee/tck/data/DataTCKExtension.java => concurrency-standalone/src/test/java/org/apache/tomee/tck/concurrency/ConcurrencyTCKExtension.java} (73%)
create mode 100644 tck/concurrency-standalone/src/test/resources/META-INF/services/org.jboss.arquillian.core.spi.LoadableExtension
(tomee) 02/05: Fix scheduled async lifecycle: stop on non-null return, reject invalid JNDI
Posted by rz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rzo1 pushed a commit to branch concurency
in repository https://gitbox.apache.org/repos/asf/tomee.git
commit a830d73543eed1c70d21607903453f58d5a8ad2c
Author: Richard Zowalla <rz...@apache.org>
AuthorDate: Thu Apr 2 14:00:19 2026 +0200
Fix scheduled async lifecycle: stop on non-null return, reject invalid JNDI
Three fixes for @Asynchronous(runAt=@Schedule(...)) TCK compliance:
1. Stop trigger loop when method returns non-null value (per spec:
"the method returns a non-null result value" ends the schedule).
Uses AtomicReference to cancel ScheduledFuture from inside Callable.
2. Throw IllegalArgumentException for invalid executor JNDI names
instead of silently falling back to default. Only default names
get the graceful fallback.
3. Add beans.xml archive processor for Concurrency TCK deployments
(OWB needs bean-discovery-mode="all" to auto-enable @Priority
interceptors).
4. Add TCK-style unit test covering CompletedResult, IncompleteFuture,
CompletedExceptionally, VoidReturn, and InvalidJNDIName scenarios.
---
.../cdi/concurrency/AsynchronousInterceptor.java | 40 ++--
...ManagedScheduledExecutorServiceImplFactory.java | 23 ++-
.../AsynchronousScheduledTCKStyleTest.java | 218 +++++++++++++++++++++
.../ConcurrencyTCKArchiveProcessor.java | 56 ++++++
.../tck/concurrency/ConcurrencyTCKExtension.java | 32 +++
...org.jboss.arquillian.core.spi.LoadableExtension | 1 +
6 files changed, 348 insertions(+), 22 deletions(-)
diff --git a/container/openejb-core/src/main/java/org/apache/openejb/cdi/concurrency/AsynchronousInterceptor.java b/container/openejb-core/src/main/java/org/apache/openejb/cdi/concurrency/AsynchronousInterceptor.java
index a62438d3d7..7a2387f4e6 100644
--- a/container/openejb-core/src/main/java/org/apache/openejb/cdi/concurrency/AsynchronousInterceptor.java
+++ b/container/openejb-core/src/main/java/org/apache/openejb/cdi/concurrency/AsynchronousInterceptor.java
@@ -40,6 +40,8 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.atomic.AtomicReference;
@Interceptor
@Asynchronous
@@ -120,42 +122,45 @@ public class AsynchronousInterceptor {
final boolean isVoid = ctx.getMethod().getReturnType() == Void.TYPE;
// A single CompletableFuture represents ALL executions in the schedule.
- // Each execution gets Asynchronous.Result.setFuture() called before ctx.proceed()
- // so the bean method can call Asynchronous.Result.getFuture() / complete().
- // The schedule stops when the future is completed, cancelled, or an exception is thrown.
+ // Per spec: "A single future represents the completion of all executions in the schedule."
+ // The schedule continues until:
+ // - the method returns a non-null result value
+ // - the method raises an exception
+ // - the future is completed (via Asynchronous.Result.complete()) or cancelled
final CompletableFuture<Object> outerFuture = mses.newIncompleteFuture();
+ final AtomicReference<ScheduledFuture<?>> scheduledRef = new AtomicReference<>();
- mses.schedule((Callable<Object>) () -> {
+ final ScheduledFuture<?> scheduledFuture = mses.schedule((Callable<Object>) () -> {
try {
Asynchronous.Result.setFuture(outerFuture);
final Object result = ctx.proceed();
if (isVoid) {
- // For void methods, the bean may call Asynchronous.Result.complete("value")
- // to signal completion. If it didn't complete the future, the schedule continues.
Asynchronous.Result.setFuture(null);
return null;
}
- if (result instanceof CompletionStage<?> cs) {
- if (result == outerFuture) {
- // Bean returned the container-provided future (via Asynchronous.Result.getFuture()).
- // It may have been completed by Asynchronous.Result.complete() inside the method.
- Asynchronous.Result.setFuture(null);
- } else {
+ // Per spec: non-null return value stops the schedule
+ if (result != null) {
+ if (result instanceof CompletionStage<?> cs && result != outerFuture) {
cs.whenComplete((val, err) -> {
if (err != null) {
outerFuture.completeExceptionally(err);
- } else if (val != null) {
+ } else {
outerFuture.complete(val);
}
Asynchronous.Result.setFuture(null);
});
}
- } else if (result != null) {
- outerFuture.complete(result);
Asynchronous.Result.setFuture(null);
+
+ // Cancel the trigger loop — method returned non-null
+ final ScheduledFuture<?> sf = scheduledRef.get();
+ if (sf != null) {
+ sf.cancel(false);
+ }
}
+ // null return: schedule continues
} catch (final Exception e) {
outerFuture.completeExceptionally(e);
Asynchronous.Result.setFuture(null);
@@ -163,6 +168,11 @@ public class AsynchronousInterceptor {
return null;
}, trigger);
+ scheduledRef.set(scheduledFuture);
+
+ // Also cancel when the future completes externally (e.g. Asynchronous.Result.complete())
+ outerFuture.whenComplete((val, err) -> scheduledFuture.cancel(false));
+
return isVoid ? null : outerFuture;
}
diff --git a/container/openejb-core/src/main/java/org/apache/openejb/resource/thread/ManagedScheduledExecutorServiceImplFactory.java b/container/openejb-core/src/main/java/org/apache/openejb/resource/thread/ManagedScheduledExecutorServiceImplFactory.java
index 44ca504da4..65d26f5e9e 100644
--- a/container/openejb-core/src/main/java/org/apache/openejb/resource/thread/ManagedScheduledExecutorServiceImplFactory.java
+++ b/container/openejb-core/src/main/java/org/apache/openejb/resource/thread/ManagedScheduledExecutorServiceImplFactory.java
@@ -38,11 +38,15 @@ public class ManagedScheduledExecutorServiceImplFactory {
private static final Logger LOGGER = Logger.getInstance(LogCategory.OPENEJB, ManagedScheduledExecutorServiceImplFactory.class);
+ private static final String DEFAULT_MES = "java:comp/DefaultManagedExecutorService";
+ private static final String DEFAULT_MSES = "java:comp/DefaultManagedScheduledExecutorService";
+
public static ManagedScheduledExecutorServiceImpl lookup(String name) {
// If the caller passes the default ManagedExecutorService JNDI name, map it to the
// default ManagedScheduledExecutorService instead
- if ("java:comp/DefaultManagedExecutorService".equals(name)) {
- name = "java:comp/DefaultManagedScheduledExecutorService";
+ final boolean isDefault = DEFAULT_MES.equals(name) || DEFAULT_MSES.equals(name);
+ if (DEFAULT_MES.equals(name)) {
+ name = DEFAULT_MSES;
}
// Try direct JNDI lookup first
@@ -58,7 +62,7 @@ public class ManagedScheduledExecutorServiceImplFactory {
// Try container JNDI with resource ID
try {
final Context ctx = SystemInstance.get().getComponent(ContainerSystem.class).getJNDIContext();
- final String resourceId = "java:comp/DefaultManagedScheduledExecutorService".equals(name)
+ final String resourceId = DEFAULT_MSES.equals(name)
? "Default Scheduled Executor Service"
: name;
@@ -67,12 +71,17 @@ public class ManagedScheduledExecutorServiceImplFactory {
return mses;
}
} catch (final NamingException ignored) {
- // fall through to default creation
+ // fall through
+ }
+
+ // Only fall back to default for the well-known default names.
+ // For custom/invalid names, throw so the caller gets RejectedExecutionException.
+ if (isDefault) {
+ LOGGER.debug("Cannot lookup ManagedScheduledExecutorService '" + name + "', creating default instance");
+ return new ManagedScheduledExecutorServiceImplFactory().create();
}
- // Graceful fallback: create a default instance
- LOGGER.debug("Cannot lookup ManagedScheduledExecutorService '" + name + "', creating default instance");
- return new ManagedScheduledExecutorServiceImplFactory().create();
+ throw new IllegalArgumentException("Cannot find ManagedScheduledExecutorService with name '" + name + "'");
}
private int core = 5;
diff --git a/container/openejb-core/src/test/java/org/apache/openejb/cdi/concurrency/AsynchronousScheduledTCKStyleTest.java b/container/openejb-core/src/test/java/org/apache/openejb/cdi/concurrency/AsynchronousScheduledTCKStyleTest.java
new file mode 100644
index 0000000000..b2a13cae21
--- /dev/null
+++ b/container/openejb-core/src/test/java/org/apache/openejb/cdi/concurrency/AsynchronousScheduledTCKStyleTest.java
@@ -0,0 +1,218 @@
+/*
+ * 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.openejb.cdi.concurrency;
+
+import jakarta.enterprise.concurrent.Asynchronous;
+import jakarta.enterprise.concurrent.Schedule;
+import jakarta.enterprise.context.RequestScoped;
+import jakarta.inject.Inject;
+import org.apache.openejb.jee.EnterpriseBean;
+import org.apache.openejb.jee.SingletonBean;
+import org.apache.openejb.junit.ApplicationComposer;
+import org.apache.openejb.testing.Module;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+/**
+ * Tests that mirror the Concurrency TCK's ReqBean pattern:
+ * - Class-level @Asynchronous (one-shot)
+ * - Method-level @Asynchronous(runAt=@Schedule(...)) (scheduled)
+ * - Uses Asynchronous.Result.getFuture() inside the bean method
+ * - Various return behaviors: NULL, COMPLETE_RESULT, COMPLETE_EXCEPTIONALLY, INCOMPLETE
+ */
+@RunWith(ApplicationComposer.class)
+public class AsynchronousScheduledTCKStyleTest {
+
+ @Inject
+ private ScheduledReqBean reqBean;
+
+ @Module
+ public EnterpriseBean ejb() {
+ return new SingletonBean(DummyEjb.class).localBean();
+ }
+
+ @Module
+ public Class<?>[] beans() {
+ return new Class<?>[]{ScheduledReqBean.class};
+ }
+
+ /**
+ * TCK: testScheduledAsynchCompletedResult
+ * Method returns null until runs==count, then completes the future via Asynchronous.Result.
+ */
+ @Test
+ public void scheduledCompletedResult() throws Exception {
+ final AtomicInteger counter = new AtomicInteger();
+ final CompletableFuture<Integer> future = reqBean.scheduledEverySecond(2, ReturnType.COMPLETE_RESULT, counter);
+
+ assertNotNull("Future should be returned by interceptor", future);
+
+ // Wait for completion — method completes the future on the 2nd invocation
+ final Integer result = future.get(15, TimeUnit.SECONDS);
+ assertEquals("Should have run exactly 2 times", Integer.valueOf(2), result);
+ }
+
+ /**
+ * TCK: testScheduledAsynchCompletedFuture
+ * Method returns INCOMPLETE future (doesn't complete it). Schedule runs once, future stays incomplete.
+ */
+ @Test
+ public void scheduledIncompleteFuture() throws Exception {
+ final AtomicInteger counter = new AtomicInteger();
+ final CompletableFuture<Integer> future = reqBean.scheduledEverySecond(1, ReturnType.INCOMPLETE, counter);
+
+ assertNotNull("Future should be returned by interceptor", future);
+
+ // The future should NOT complete — it's INCOMPLETE
+ try {
+ future.get(3, TimeUnit.SECONDS);
+ fail("Should have timed out — future is incomplete");
+ } catch (final TimeoutException e) {
+ // expected
+ }
+
+ assertFalse("Future should not be done", future.isDone());
+ assertFalse("Future should not be cancelled", future.isCancelled());
+
+ // Should have executed exactly once
+ assertEquals("Schedule should have executed exactly once", 1, counter.get());
+
+ future.cancel(true);
+ }
+
+ /**
+ * TCK: testScheduledAsynchCompletedExceptionally
+ * Method completes the future exceptionally.
+ */
+ @Test
+ public void scheduledCompletedExceptionally() throws Exception {
+ final AtomicInteger counter = new AtomicInteger();
+ final CompletableFuture<Integer> future = reqBean.scheduledEverySecond(1, ReturnType.COMPLETE_EXCEPTIONALLY, counter);
+
+ assertNotNull("Future should be returned by interceptor", future);
+
+ try {
+ future.get(15, TimeUnit.SECONDS);
+ fail("Should have completed exceptionally");
+ } catch (final Exception e) {
+ assertTrue("Future should be done", future.isDone());
+ assertTrue("Future should be completed exceptionally", future.isCompletedExceptionally());
+ }
+ }
+
+ /**
+ * TCK: testScheduledAsynchVoidReturn
+ * Void method with @Asynchronous(runAt=...) — should execute repeatedly.
+ */
+ @Test
+ public void scheduledVoidReturn() throws Exception {
+ final AtomicInteger counter = new AtomicInteger();
+ reqBean.scheduledVoidEverySecond(3, counter);
+
+ // Wait for 3 executions
+ final long start = System.currentTimeMillis();
+ while (counter.get() < 3 && System.currentTimeMillis() - start < 15000) {
+ Thread.sleep(200);
+ }
+ assertTrue("Void method should have executed at least 3 times, got: " + counter.get(),
+ counter.get() >= 3);
+ }
+
+ /**
+ * TCK: testScheduledAsynchWithInvalidJNDIName
+ * Method with invalid executor JNDI name should throw RejectedExecutionException.
+ */
+ @Test
+ public void scheduledWithInvalidJNDIName() throws Exception {
+ try {
+ reqBean.scheduledInvalidExecutor();
+ fail("Should have thrown an exception for invalid executor");
+ } catch (final jakarta.ejb.EJBException | java.util.concurrent.RejectedExecutionException e) {
+ // expected — invalid JNDI name
+ }
+ }
+
+ // --- Bean ---
+
+ public enum ReturnType {
+ NULL, COMPLETE_RESULT, COMPLETE_EXCEPTIONALLY, INCOMPLETE, THROW_EXCEPTION
+ }
+
+ @Asynchronous
+ @RequestScoped
+ public static class ScheduledReqBean {
+
+ @Asynchronous(runAt = @Schedule(cron = "* * * * * *"))
+ public CompletableFuture<Integer> scheduledEverySecond(final int runs, final ReturnType type,
+ final AtomicInteger counter) {
+ final int count = counter.incrementAndGet();
+
+ // Return null until we've reached the target number of runs
+ if (runs != count) {
+ return null;
+ }
+
+ final CompletableFuture<Integer> future = Asynchronous.Result.getFuture();
+
+ switch (type) {
+ case NULL:
+ return null;
+ case COMPLETE_EXCEPTIONALLY:
+ future.completeExceptionally(new Exception("Expected exception"));
+ return future;
+ case COMPLETE_RESULT:
+ future.complete(count);
+ return future;
+ case INCOMPLETE:
+ return future; // don't complete it
+ case THROW_EXCEPTION:
+ throw new RuntimeException("Expected exception");
+ default:
+ return null;
+ }
+ }
+
+ @Asynchronous(runAt = @Schedule(cron = "* * * * * *"))
+ public void scheduledVoidEverySecond(final int runs, final AtomicInteger counter) {
+ final int count = counter.incrementAndGet();
+ if (count >= runs) {
+ Asynchronous.Result.getFuture().complete(null);
+ }
+ }
+
+ @Asynchronous(executor = "java:comp/env/invalid/executor",
+ runAt = @Schedule(cron = "* * * * * *"))
+ public CompletableFuture<String> scheduledInvalidExecutor() {
+ throw new UnsupportedOperationException("Should not reach here with invalid executor");
+ }
+ }
+
+ @jakarta.ejb.Singleton
+ public static class DummyEjb {
+ }
+}
diff --git a/tck/concurrency-standalone/src/test/java/org/apache/tomee/tck/concurrency/ConcurrencyTCKArchiveProcessor.java b/tck/concurrency-standalone/src/test/java/org/apache/tomee/tck/concurrency/ConcurrencyTCKArchiveProcessor.java
new file mode 100644
index 0000000000..eba62c43a5
--- /dev/null
+++ b/tck/concurrency-standalone/src/test/java/org/apache/tomee/tck/concurrency/ConcurrencyTCKArchiveProcessor.java
@@ -0,0 +1,56 @@
+/*
+ * 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.tomee.tck.concurrency;
+
+import org.jboss.arquillian.container.test.spi.client.deployment.ApplicationArchiveProcessor;
+import org.jboss.arquillian.test.spi.TestClass;
+import org.jboss.shrinkwrap.api.Archive;
+import org.jboss.shrinkwrap.api.asset.StringAsset;
+import org.jboss.shrinkwrap.api.spec.WebArchive;
+
+/**
+ * Arquillian ApplicationArchiveProcessor that adds a beans.xml with
+ * {@code bean-discovery-mode="all"} to Concurrency TCK deployments.
+ *
+ * <p>This is needed because OWB (OpenWebBeans) in TomEE does not yet
+ * auto-enable {@code @Priority} interceptors without an explicit beans.xml.
+ * The {@code AsynchronousInterceptor} relies on this to be activated
+ * in deployed WARs.</p>
+ */
+public class ConcurrencyTCKArchiveProcessor implements ApplicationArchiveProcessor {
+
+ private static final String BEANS_XML =
+ "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n" +
+ "<beans xmlns=\"https://jakarta.ee/xml/ns/jakartaee\"\n" +
+ " xmlns:xsi=\"http://www.w3.org/2001/XMLSchema-instance\"\n" +
+ " xsi:schemaLocation=\"https://jakarta.ee/xml/ns/jakartaee\n" +
+ " https://jakarta.ee/xml/ns/jakartaee/beans_4_0.xsd\"\n" +
+ " version=\"4.0\"\n" +
+ " bean-discovery-mode=\"all\">\n" +
+ "</beans>\n";
+
+ @Override
+ public void process(final Archive<?> archive, final TestClass testClass) {
+ if (archive instanceof WebArchive) {
+ final WebArchive war = (WebArchive) archive;
+
+ if (!archive.contains("WEB-INF/beans.xml")) {
+ war.addAsWebInfResource(new StringAsset(BEANS_XML), "beans.xml");
+ }
+ }
+ }
+}
diff --git a/tck/concurrency-standalone/src/test/java/org/apache/tomee/tck/concurrency/ConcurrencyTCKExtension.java b/tck/concurrency-standalone/src/test/java/org/apache/tomee/tck/concurrency/ConcurrencyTCKExtension.java
new file mode 100644
index 0000000000..2fc4d1df11
--- /dev/null
+++ b/tck/concurrency-standalone/src/test/java/org/apache/tomee/tck/concurrency/ConcurrencyTCKExtension.java
@@ -0,0 +1,32 @@
+/*
+ * 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.tomee.tck.concurrency;
+
+import org.jboss.arquillian.container.test.spi.client.deployment.ApplicationArchiveProcessor;
+import org.jboss.arquillian.core.spi.LoadableExtension;
+
+/**
+ * Arquillian LoadableExtension that registers the ConcurrencyTCKArchiveProcessor.
+ * Discovered via META-INF/services/org.jboss.arquillian.core.spi.LoadableExtension.
+ */
+public class ConcurrencyTCKExtension implements LoadableExtension {
+
+ @Override
+ public void register(final ExtensionBuilder builder) {
+ builder.service(ApplicationArchiveProcessor.class, ConcurrencyTCKArchiveProcessor.class);
+ }
+}
diff --git a/tck/concurrency-standalone/src/test/resources/META-INF/services/org.jboss.arquillian.core.spi.LoadableExtension b/tck/concurrency-standalone/src/test/resources/META-INF/services/org.jboss.arquillian.core.spi.LoadableExtension
new file mode 100644
index 0000000000..d7e5ba2411
--- /dev/null
+++ b/tck/concurrency-standalone/src/test/resources/META-INF/services/org.jboss.arquillian.core.spi.LoadableExtension
@@ -0,0 +1 @@
+org.apache.tomee.tck.concurrency.ConcurrencyTCKExtension
(tomee) 01/05: Fix scheduled async interceptor to call setFuture before ctx.proceed
Posted by rz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rzo1 pushed a commit to branch concurency
in repository https://gitbox.apache.org/repos/asf/tomee.git
commit bdc3fe93952651f6d1ce5be0203aed0c8372295a
Author: Richard Zowalla <rz...@apache.org>
AuthorDate: Thu Apr 2 13:13:38 2026 +0200
Fix scheduled async interceptor to call setFuture before ctx.proceed
The TCK beans call Asynchronous.Result.getFuture() inside scheduled
methods. The interceptor must call setFuture() before ctx.proceed()
for both void and non-void return types, otherwise getFuture() throws
IllegalStateException. Use Callable path for all scheduled methods
to ensure proper future lifecycle.
---
.../cdi/concurrency/AsynchronousInterceptor.java | 48 ++++++++++++----------
1 file changed, 26 insertions(+), 22 deletions(-)
diff --git a/container/openejb-core/src/main/java/org/apache/openejb/cdi/concurrency/AsynchronousInterceptor.java b/container/openejb-core/src/main/java/org/apache/openejb/cdi/concurrency/AsynchronousInterceptor.java
index d896e9eaf2..a62438d3d7 100644
--- a/container/openejb-core/src/main/java/org/apache/openejb/cdi/concurrency/AsynchronousInterceptor.java
+++ b/container/openejb-core/src/main/java/org/apache/openejb/cdi/concurrency/AsynchronousInterceptor.java
@@ -119,19 +119,10 @@ public class AsynchronousInterceptor {
final ZonedTrigger trigger = ScheduleHelper.toTrigger(schedules);
final boolean isVoid = ctx.getMethod().getReturnType() == Void.TYPE;
- if (isVoid) {
- // void method: schedule as Runnable, runs indefinitely until cancelled
- mses.schedule((Runnable) () -> {
- try {
- ctx.proceed();
- } catch (final Exception e) {
- LOGGER.warning("Scheduled async method threw exception", e);
- }
- }, trigger);
- return null;
- }
-
- // non-void: schedule as Callable, each invocation gets a fresh future via Asynchronous.Result
+ // A single CompletableFuture represents ALL executions in the schedule.
+ // Each execution gets Asynchronous.Result.setFuture() called before ctx.proceed()
+ // so the bean method can call Asynchronous.Result.getFuture() / complete().
+ // The schedule stops when the future is completed, cancelled, or an exception is thrown.
final CompletableFuture<Object> outerFuture = mses.newIncompleteFuture();
mses.schedule((Callable<Object>) () -> {
@@ -139,16 +130,29 @@ public class AsynchronousInterceptor {
Asynchronous.Result.setFuture(outerFuture);
final Object result = ctx.proceed();
+ if (isVoid) {
+ // For void methods, the bean may call Asynchronous.Result.complete("value")
+ // to signal completion. If it didn't complete the future, the schedule continues.
+ Asynchronous.Result.setFuture(null);
+ return null;
+ }
+
if (result instanceof CompletionStage<?> cs) {
- cs.whenComplete((val, err) -> {
- if (err != null) {
- outerFuture.completeExceptionally(err);
- } else if (val != null) {
- outerFuture.complete(val);
- }
+ if (result == outerFuture) {
+ // Bean returned the container-provided future (via Asynchronous.Result.getFuture()).
+ // It may have been completed by Asynchronous.Result.complete() inside the method.
Asynchronous.Result.setFuture(null);
- });
- } else if (result != null && result != outerFuture) {
+ } else {
+ cs.whenComplete((val, err) -> {
+ if (err != null) {
+ outerFuture.completeExceptionally(err);
+ } else if (val != null) {
+ outerFuture.complete(val);
+ }
+ Asynchronous.Result.setFuture(null);
+ });
+ }
+ } else if (result != null) {
outerFuture.complete(result);
Asynchronous.Result.setFuture(null);
}
@@ -159,7 +163,7 @@ public class AsynchronousInterceptor {
return null;
}, trigger);
- return outerFuture;
+ return isVoid ? null : outerFuture;
}
private Exception validate(final Method method) {
(tomee) 03/05: Use bean-discovery-mode=annotated in Concurrency TCK archive processor
Posted by rz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rzo1 pushed a commit to branch concurency
in repository https://gitbox.apache.org/repos/asf/tomee.git
commit 8447b7f17d5a8a1c6533d45489fe166e18939730
Author: Richard Zowalla <rz...@apache.org>
AuthorDate: Thu Apr 2 14:43:37 2026 +0200
Use bean-discovery-mode=annotated in Concurrency TCK archive processor
bean-discovery-mode=all caused deployment failures for some TCK WARs
by scanning too many classes. Switch to annotated mode which is the
CDI 4.0 default and sufficient for discovering annotated beans.
---
.../apache/tomee/tck/concurrency/ConcurrencyTCKArchiveProcessor.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/tck/concurrency-standalone/src/test/java/org/apache/tomee/tck/concurrency/ConcurrencyTCKArchiveProcessor.java b/tck/concurrency-standalone/src/test/java/org/apache/tomee/tck/concurrency/ConcurrencyTCKArchiveProcessor.java
index eba62c43a5..e7870778ad 100644
--- a/tck/concurrency-standalone/src/test/java/org/apache/tomee/tck/concurrency/ConcurrencyTCKArchiveProcessor.java
+++ b/tck/concurrency-standalone/src/test/java/org/apache/tomee/tck/concurrency/ConcurrencyTCKArchiveProcessor.java
@@ -40,7 +40,7 @@ public class ConcurrencyTCKArchiveProcessor implements ApplicationArchiveProcess
" xsi:schemaLocation=\"https://jakarta.ee/xml/ns/jakartaee\n" +
" https://jakarta.ee/xml/ns/jakartaee/beans_4_0.xsd\"\n" +
" version=\"4.0\"\n" +
- " bean-discovery-mode=\"all\">\n" +
+ " bean-discovery-mode=\"annotated\">\n" +
"</beans>\n";
@Override
(tomee) 05/05: Fix JNDI lookup for java:module/ and java:app/ scoped scheduled executors
Posted by rz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rzo1 pushed a commit to branch concurency
in repository https://gitbox.apache.org/repos/asf/tomee.git
commit 40716ae89544daaff27f51c31f38f1fd2e419d96
Author: Richard Zowalla <rz...@apache.org>
AuthorDate: Thu Apr 2 14:58:45 2026 +0200
Fix JNDI lookup for java:module/ and java:app/ scoped scheduled executors
Strip java: prefix in ManagedScheduledExecutorServiceImplFactory.lookup()
fallback path to match how resources are registered via cleanUpName().
Without this, java:module/concurrent/ScheduledExecutorB lookups fail
because the resource is bound as module/concurrent/ScheduledExecutorB.
Add Arquillian test verifying both java:module/ and java:app/ scoped
@ManagedScheduledExecutorDefinition work with scheduled async methods.
---
.../ScheduledAsyncCustomExecutorTest.java | 110 +++++++++++++++++++++
...ManagedScheduledExecutorServiceImplFactory.java | 10 +-
2 files changed, 117 insertions(+), 3 deletions(-)
diff --git a/arquillian/arquillian-tomee-tests/arquillian-tomee-webprofile-tests/src/test/java/org/apache/openejb/arquillian/tests/concurrency/ScheduledAsyncCustomExecutorTest.java b/arquillian/arquillian-tomee-tests/arquillian-tomee-webprofile-tests/src/test/java/org/apache/openejb/arquillian/tests/concurrency/ScheduledAsyncCustomExecutorTest.java
new file mode 100644
index 0000000000..aa728c4e90
--- /dev/null
+++ b/arquillian/arquillian-tomee-tests/arquillian-tomee-webprofile-tests/src/test/java/org/apache/openejb/arquillian/tests/concurrency/ScheduledAsyncCustomExecutorTest.java
@@ -0,0 +1,110 @@
+/*
+ * 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.openejb.arquillian.tests.concurrency;
+
+import jakarta.enterprise.concurrent.Asynchronous;
+import jakarta.enterprise.concurrent.ManagedScheduledExecutorDefinition;
+import jakarta.enterprise.concurrent.Schedule;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Inject;
+import org.jboss.arquillian.container.test.api.Deployment;
+import org.jboss.arquillian.junit.Arquillian;
+import org.jboss.shrinkwrap.api.ArchivePaths;
+import org.jboss.shrinkwrap.api.ShrinkWrap;
+import org.jboss.shrinkwrap.api.asset.EmptyAsset;
+import org.jboss.shrinkwrap.api.spec.WebArchive;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
+/**
+ * Arquillian test that mirrors the TCK pattern of using
+ * {@code @ManagedScheduledExecutorDefinition} with a custom JNDI name
+ * and {@code @Asynchronous(executor="java:module/...", runAt=@Schedule(...))}.
+ *
+ * <p>This verifies that {@code java:module/} and {@code java:app/} scoped
+ * executor lookups work for scheduled async methods.</p>
+ */
+@RunWith(Arquillian.class)
+public class ScheduledAsyncCustomExecutorTest {
+
+ @Inject
+ private ScheduledBeanWithCustomExecutor bean;
+
+ @Deployment
+ public static WebArchive createDeployment() {
+ return ShrinkWrap.create(WebArchive.class, "ScheduledAsyncCustomExecutorTest.war")
+ .addClasses(ScheduledBeanWithCustomExecutor.class)
+ .addAsWebInfResource(EmptyAsset.INSTANCE, ArchivePaths.create("beans.xml"));
+ }
+
+ @Test
+ public void scheduledWithModuleScopedExecutor() throws Exception {
+ final AtomicInteger counter = new AtomicInteger();
+ final CompletableFuture<Integer> future = bean.scheduledWithModuleExecutor(2, counter);
+
+ assertNotNull("Future should be returned", future);
+ final Integer result = future.get(15, TimeUnit.SECONDS);
+ assertEquals("Should complete after 2 runs", Integer.valueOf(2), result);
+ }
+
+ @Test
+ public void scheduledWithAppScopedExecutor() throws Exception {
+ final AtomicInteger counter = new AtomicInteger();
+ final CompletableFuture<Integer> future = bean.scheduledWithAppExecutor(1, counter);
+
+ assertNotNull("Future should be returned", future);
+ final Integer result = future.get(15, TimeUnit.SECONDS);
+ assertEquals("Should complete after 1 run", Integer.valueOf(1), result);
+ }
+
+ @ManagedScheduledExecutorDefinition(name = "java:module/concurrent/TestScheduledExecutor")
+ @ManagedScheduledExecutorDefinition(name = "java:app/concurrent/TestAppScheduledExecutor")
+ @ApplicationScoped
+ public static class ScheduledBeanWithCustomExecutor {
+
+ @Asynchronous(executor = "java:module/concurrent/TestScheduledExecutor",
+ runAt = @Schedule(cron = "* * * * * *"))
+ public CompletableFuture<Integer> scheduledWithModuleExecutor(final int runs, final AtomicInteger counter) {
+ final int count = counter.incrementAndGet();
+ if (count < runs) {
+ return null;
+ }
+ final CompletableFuture<Integer> future = Asynchronous.Result.getFuture();
+ future.complete(count);
+ return future;
+ }
+
+ @Asynchronous(executor = "java:app/concurrent/TestAppScheduledExecutor",
+ runAt = @Schedule(cron = "* * * * * *"))
+ public CompletableFuture<Integer> scheduledWithAppExecutor(final int runs, final AtomicInteger counter) {
+ final int count = counter.incrementAndGet();
+ if (count < runs) {
+ return null;
+ }
+ final CompletableFuture<Integer> future = Asynchronous.Result.getFuture();
+ future.complete(count);
+ return future;
+ }
+ }
+}
diff --git a/container/openejb-core/src/main/java/org/apache/openejb/resource/thread/ManagedScheduledExecutorServiceImplFactory.java b/container/openejb-core/src/main/java/org/apache/openejb/resource/thread/ManagedScheduledExecutorServiceImplFactory.java
index 65d26f5e9e..6395511848 100644
--- a/container/openejb-core/src/main/java/org/apache/openejb/resource/thread/ManagedScheduledExecutorServiceImplFactory.java
+++ b/container/openejb-core/src/main/java/org/apache/openejb/resource/thread/ManagedScheduledExecutorServiceImplFactory.java
@@ -62,9 +62,13 @@ public class ManagedScheduledExecutorServiceImplFactory {
// Try container JNDI with resource ID
try {
final Context ctx = SystemInstance.get().getComponent(ContainerSystem.class).getJNDIContext();
- final String resourceId = DEFAULT_MSES.equals(name)
- ? "Default Scheduled Executor Service"
- : name;
+ String resourceId;
+ if (DEFAULT_MSES.equals(name)) {
+ resourceId = "Default Scheduled Executor Service";
+ } else {
+ // Strip java: prefix to match how resources are registered (via cleanUpName)
+ resourceId = name.startsWith("java:") ? name.substring("java:".length()) : name;
+ }
final Object obj = ctx.lookup("openejb/Resource/" + resourceId);
if (obj instanceof ManagedScheduledExecutorServiceImpl mses) {
(tomee) 04/05: Add multiple schedules and maxAsync unit tests for scheduled async
Posted by rz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rzo1 pushed a commit to branch concurency
in repository https://gitbox.apache.org/repos/asf/tomee.git
commit 020017b41d25fe3ef522942a2e969c3502ee9424
Author: Richard Zowalla <rz...@apache.org>
AuthorDate: Thu Apr 2 14:48:16 2026 +0200
Add multiple schedules and maxAsync unit tests for scheduled async
Extend TCK-style test coverage with multipleSchedules (composite
trigger with two @Schedule annotations) and ignoresMaxAsync (concurrent
scheduled method invocations). Both pass — remaining TCK failures for
these tests are JNDI scoping issues with java:module/ executor names.
---
.../AsynchronousScheduledTCKStyleTest.java | 52 ++++++++++++++++++++++
1 file changed, 52 insertions(+)
diff --git a/container/openejb-core/src/test/java/org/apache/openejb/cdi/concurrency/AsynchronousScheduledTCKStyleTest.java b/container/openejb-core/src/test/java/org/apache/openejb/cdi/concurrency/AsynchronousScheduledTCKStyleTest.java
index b2a13cae21..b6a5f7555e 100644
--- a/container/openejb-core/src/test/java/org/apache/openejb/cdi/concurrency/AsynchronousScheduledTCKStyleTest.java
+++ b/container/openejb-core/src/test/java/org/apache/openejb/cdi/concurrency/AsynchronousScheduledTCKStyleTest.java
@@ -157,6 +157,44 @@ public class AsynchronousScheduledTCKStyleTest {
}
}
+ /**
+ * TCK: testScheduledAsynchWithMultipleSchedules
+ * Method with two @Schedule annotations — should use composite trigger.
+ */
+ @Test
+ public void scheduledWithMultipleSchedules() throws Exception {
+ final AtomicInteger counter = new AtomicInteger();
+ final CompletableFuture<String> future = reqBean.scheduledMultipleSchedules(1, counter);
+
+ assertNotNull("Future should be returned by interceptor", future);
+
+ final String result = future.get(15, TimeUnit.SECONDS);
+ assertNotNull("Should have completed with a result", result);
+ assertEquals("Should have run exactly once", 1, counter.get());
+ }
+
+ /**
+ * TCK: testScheduledAsynchIgnoresMaxAsync
+ * Multiple concurrent scheduled async method invocations should all run regardless of maxAsync.
+ * Here we just verify that the scheduled method works when called multiple times.
+ */
+ @Test
+ public void scheduledIgnoresMaxAsync() throws Exception {
+ final AtomicInteger counter1 = new AtomicInteger();
+ final AtomicInteger counter2 = new AtomicInteger();
+ final CompletableFuture<Integer> future1 = reqBean.scheduledEverySecond(3, ReturnType.COMPLETE_RESULT, counter1);
+ final CompletableFuture<Integer> future2 = reqBean.scheduledEverySecond(3, ReturnType.COMPLETE_RESULT, counter2);
+
+ assertNotNull("First future should not be null", future1);
+ assertNotNull("Second future should not be null", future2);
+
+ // Both should complete
+ final Integer result1 = future1.get(15, TimeUnit.SECONDS);
+ final Integer result2 = future2.get(15, TimeUnit.SECONDS);
+ assertEquals(Integer.valueOf(3), result1);
+ assertEquals(Integer.valueOf(3), result2);
+ }
+
// --- Bean ---
public enum ReturnType {
@@ -205,6 +243,20 @@ public class AsynchronousScheduledTCKStyleTest {
}
}
+ @Asynchronous(runAt = {
+ @Schedule(cron = "* * * * * *"),
+ @Schedule(cron = "*/2 * * * * *")
+ })
+ public CompletableFuture<String> scheduledMultipleSchedules(final int runs, final AtomicInteger counter) {
+ final int count = counter.incrementAndGet();
+ if (runs != count) {
+ return null;
+ }
+ final CompletableFuture<String> future = Asynchronous.Result.getFuture();
+ future.complete("completed-" + count);
+ return future;
+ }
+
@Asynchronous(executor = "java:comp/env/invalid/executor",
runAt = @Schedule(cron = "* * * * * *"))
public CompletableFuture<String> scheduledInvalidExecutor() {