Isolate scheduled @asynchronous firings from maxAsync

Add a secondary ScheduledExecutorService to every
ManagedScheduledExecutorServiceImpl, dedicated to dispatching
@asynchronous(runAt=@schedule(...)) firings. The secondary pool shares
the primary's ManagedThreadFactory (preserving naming, virtual flag,
priority, and context) but is sized independently via a new
ScheduledAsyncCore resource property (default 5). The interceptor now
triggers on the secondary pool, so scheduled firings are no longer
throttled by the primary pool's corePoolSize=maxAsync limit — satisfying
Concurrency 3.1 §3.1 "Scheduled asynchronous methods are treated similar
to other scheduled tasks in that they are not subject to max-async
constraints" without losing the clause from PR #2577 review that firings
must run on the executor named in @asynchronous(executor=).

The plain-MES fallback wrapper borrows the default MSES's secondary so
short-lived per-invocation wrappers do not leak a fresh pool each time;
an ownership flag on the 4-arg constructor controls which pools
destroyResource() shuts down.

Also fix a pre-existing race in CUTriggerScheduledFuture: cancel() can
race ahead of the recursive scheduleNextRun(), leaving futureRef
pointing at the just-completed execution so the underlying future
reports done rather than cancelled. isCancelled() now falls back to the
TriggerTask's cancelled flag.

Tests:
- ScheduledAsyncCustomFactoryTest — scheduled firing runs on a
  ManageableThread from the custom MSES's factory, not the primary pool.
- ManagedScheduledExecutorSecondaryPoolLifecycleTest — owned secondary
  is shut down on destroyResource(); borrowed secondary is not.

TCK Web profile back to 196/0/0/14.

Author: rzo1

container/openejb-core/src/main/java/org/apache/openejb/cdi/concurrency/AsynchronousInterceptor.java

@@ -135,8 +135,9 @@ private Object aroundInvokeScheduled(final InvocationContext ctx, final Asynchro
         final ContextServiceImpl.Snapshot snapshot = ctxService.snapshot(null);
 
         // Run scheduled firings on the requested executor so the user's thread factory,
-        // priorities, and virtual-thread settings apply.
-        final ScheduledExecutorService triggerDelegate = mses.getDelegate();
+        // priorities, and virtual-thread settings apply. Use the secondary pool so firings
+        // are not throttled by maxAsync (per Concurrency 3.1 §3.1).
+        final ScheduledExecutorService triggerDelegate = mses.getScheduledAsyncDelegate();
 
         // A single CompletableFuture represents ALL executions in the schedule.
         // Per spec: "A single future represents the completion of all executions in the schedule."
@@ -177,7 +178,10 @@ private ManagedScheduledExecutorServiceImpl resolveMses(final String executorNam
                 final ContextServiceImpl mesContextService = (ContextServiceImpl) plainMes.getContextService();
                 final ManagedScheduledExecutorServiceImpl defaultMses =
                         ManagedScheduledExecutorServiceImplFactory.lookup("java:comp/DefaultManagedScheduledExecutorService");
-                return new ManagedScheduledExecutorServiceImpl(defaultMses.getDelegate(), mesContextService);
+                // Borrow the default MSES's secondary pool so this short-lived wrapper does not
+                // leak a fresh ScheduledThreadPoolExecutor per invocation. ownsScheduledAsyncDelegate=false.
+                return new ManagedScheduledExecutorServiceImpl(defaultMses.getDelegate(), mesContextService,
+                        defaultMses.getScheduledAsyncDelegate(), false);
             } catch (final Exception fallbackEx) {
                 throw new RejectedExecutionException("Cannot lookup executor for scheduled async method", e);
             }

container/openejb-core/src/main/java/org/apache/openejb/resource/thread/ManagedScheduledExecutorServiceImplFactory.java

@@ -89,19 +89,23 @@ public static ManagedScheduledExecutorServiceImpl lookup(String name) {
     }
 
     private int core = 5;
+    private int scheduledAsyncCore = 5;
     private String threadFactory = ManagedThreadFactoryImpl.class.getName();
     private boolean virtual;
 
     private String context;
 
     public ManagedScheduledExecutorServiceImpl create(final ContextServiceImpl contextService) {
-        return new ManagedScheduledExecutorServiceImpl(createScheduledExecutorService(), contextService);
+        final Pools pools = createScheduledExecutorServicePools();
+        return new ManagedScheduledExecutorServiceImpl(pools.primary, contextService, pools.secondary, true);
     }
     public ManagedScheduledExecutorServiceImpl create() {
-        return new ManagedScheduledExecutorServiceImpl(createScheduledExecutorService(), ContextServiceImplFactory.lookupOrDefault(context));
+        final Pools pools = createScheduledExecutorServicePools();
+        return new ManagedScheduledExecutorServiceImpl(pools.primary, ContextServiceImplFactory.lookupOrDefault(context),
+                pools.secondary, true);
     }
 
-    private ScheduledExecutorService createScheduledExecutorService() {
+    private Pools createScheduledExecutorServicePools() {
         ManagedThreadFactory managedThreadFactory;
         try {
             // For the default factory, bypass reflective instantiation so the configured
@@ -114,13 +118,28 @@ private ScheduledExecutorService createScheduledExecutorService() {
             managedThreadFactory = new ManagedThreadFactoryImpl(ManagedThreadFactoryImpl.DEFAULT_PREFIX, null, ContextServiceImplFactory.lookupOrDefault(context), virtual);
         }
 
-        return new ScheduledThreadPoolExecutor(core, managedThreadFactory, CURejectHandler.INSTANCE);
+        // Primary pool — regular async submissions; corePoolSize follows maxAsync.
+        final ScheduledExecutorService primary =
+                new ScheduledThreadPoolExecutor(core, managedThreadFactory, CURejectHandler.INSTANCE);
+        // Secondary pool — scheduled @Asynchronous firings. Per Concurrency 3.1 §3.1 these
+        // must NOT be throttled by maxAsync, so size the pool independently while reusing
+        // the same stateless thread factory (preserves naming / virtual / priority).
+        final ScheduledExecutorService secondary =
+                new ScheduledThreadPoolExecutor(scheduledAsyncCore, managedThreadFactory, CURejectHandler.INSTANCE);
+        return new Pools(primary, secondary);
     }
 
     public void setCore(final int core) {
         this.core = core;
     }
 
+    public void setScheduledAsyncCore(final int scheduledAsyncCore) {
+        this.scheduledAsyncCore = scheduledAsyncCore;
+    }
+
+    private record Pools(ScheduledExecutorService primary, ScheduledExecutorService secondary) {
+    }
+
     public void setThreadFactory(final String threadFactory) {
         this.threadFactory = threadFactory;
     }

container/openejb-core/src/main/java/org/apache/openejb/threads/future/CUTriggerScheduledFuture.java

@@ -44,6 +44,17 @@ public boolean cancel(boolean mayInterruptIfRunning) {
         return super.cancel(mayInterruptIfRunning);
     }
 
+    @Override
+    public boolean isCancelled() {
+        // Also honour the TriggerTask's cancelled flag. The underlying delegate future
+        // reachable through the facade can point at the just-completed execution if
+        // cancel() raced ahead of the recursive scheduleNextRun(); in that case cancel()
+        // set cancelled=true on the TriggerTask but the delegate future reports done
+        // rather than cancelled. Without this override, isCancelled() would return false
+        // for a future the caller explicitly cancelled.
+        return super.isCancelled() || ((TriggerTask<V>) listener).isCancelled();
+    }
+
     @Override
     public V get() throws InterruptedException, ExecutionException {
         V result = super.get();

container/openejb-core/src/main/java/org/apache/openejb/threads/impl/ManagedScheduledExecutorServiceImpl.java

@@ -27,11 +27,14 @@
 import jakarta.enterprise.concurrent.ManagedScheduledExecutorService;
 import jakarta.enterprise.concurrent.ManagedTask;
 import jakarta.enterprise.concurrent.Trigger;
+import org.apache.openejb.util.LogCategory;
+import org.apache.openejb.util.Logger;
 import java.lang.reflect.InvocationHandler;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.lang.reflect.Proxy;
 import java.util.Date;
+import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.Callable;
@@ -42,13 +45,25 @@
 import java.util.concurrent.atomic.AtomicReference;
 
 public class ManagedScheduledExecutorServiceImpl extends ManagedExecutorServiceImpl implements ManagedScheduledExecutorService {
+    private static final Logger LOGGER = Logger.getInstance(LogCategory.OPENEJB, ManagedScheduledExecutorServiceImpl.class);
+
     private final ScheduledExecutorService delegate;
     private final ContextServiceImpl contextService;
+    private final ScheduledExecutorService scheduledAsyncDelegate;
+    private final boolean ownsScheduledAsyncDelegate;
 
     public ManagedScheduledExecutorServiceImpl(final ScheduledExecutorService delegate, final ContextServiceImpl contextService) {
+        this(delegate, contextService, delegate, false);
+    }
+
+    public ManagedScheduledExecutorServiceImpl(final ScheduledExecutorService delegate, final ContextServiceImpl contextService,
+                                               final ScheduledExecutorService scheduledAsyncDelegate,
+                                               final boolean ownsScheduledAsyncDelegate) {
         super(delegate, contextService);
         this.delegate = delegate;
         this.contextService = contextService;
+        this.scheduledAsyncDelegate = scheduledAsyncDelegate != null ? scheduledAsyncDelegate : delegate;
+        this.ownsScheduledAsyncDelegate = ownsScheduledAsyncDelegate;
     }
 
 
@@ -137,6 +152,35 @@ public ScheduledExecutorService getDelegate() {
         return delegate;
     }
 
+    /**
+     * Secondary scheduling pool used to dispatch {@code @Asynchronous(runAt=@Schedule(...))}
+     * firings. Per Jakarta Concurrency 3.1 §3.1, scheduled asynchronous methods are not
+     * subject to {@code max-async}, so firings must not queue behind regular async work
+     * occupying the primary delegate's core threads.
+     */
+    public ScheduledExecutorService getScheduledAsyncDelegate() {
+        return scheduledAsyncDelegate;
+    }
+
+    @Override
+    public void destroyResource() {
+        if (ownsScheduledAsyncDelegate && scheduledAsyncDelegate != null && scheduledAsyncDelegate != delegate) {
+            final List<Runnable> leftover = scheduledAsyncDelegate.shutdownNow();
+            if (!leftover.isEmpty()) {
+                LOGGER.warning(leftover.size() + " scheduled-async tasks to execute");
+                for (final Runnable runnable : leftover) {
+                    try {
+                        LOGGER.info("Executing " + runnable);
+                        runnable.run();
+                    } catch (final Throwable th) {
+                        LOGGER.error(th.getMessage(), th);
+                    }
+                }
+            }
+        }
+        super.destroyResource();
+    }
+
     /**
      * Automatically resolves an AtomicReference
      * @param delegate

container/openejb-core/src/main/java/org/apache/openejb/threads/task/TriggerTask.java

@@ -142,6 +142,10 @@ public void cancelScheduling() {
         }
     }
 
+    public boolean isCancelled() {
+        return cancelled;
+    }
+
     private static class LastExecutionImpl implements LastExecution {
         private final String identityName;
         private final Object result;

container/openejb-core/src/test/java/org/apache/openejb/cdi/concurrency/ScheduledAsyncCustomFactoryTest.java

@@ -0,0 +1,147 @@
+/*
+ * 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.ManageableThread;
+import jakarta.enterprise.concurrent.ManagedScheduledExecutorDefinition;
+import jakarta.enterprise.concurrent.ManagedScheduledExecutorService;
+import jakarta.enterprise.concurrent.Schedule;
+import jakarta.enterprise.context.ApplicationScoped;
+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.apache.openejb.threads.impl.ManagedThreadFactoryImpl;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import javax.naming.InitialContext;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Verifies that scheduled {@code @Asynchronous} firings honor the requested
+ * {@link ManagedScheduledExecutorService}'s thread factory: the firing thread
+ * is produced by the same {@link ManagedThreadFactoryImpl} as the primary pool
+ * (same naming prefix, {@link ManageableThread} shape), not a stranger pool such
+ * as the default MSES. Locks down clause A of Concurrency 3.1 §3.1 (firings run
+ * with the requested executor's thread factory / virtual / priority).
+ */
+@RunWith(ApplicationComposer.class)
+public class ScheduledAsyncCustomFactoryTest {
+
+    private static final String MSES_JNDI = "java:module/custom/factoryMSES";
+
+    @Inject
+    private FactoryBean bean;
+
+    @Module
+    public EnterpriseBean ejb() {
+        return new SingletonBean(DummyEjb.class).localBean();
+    }
+
+    @Module
+    public Class<?>[] beans() {
+        return new Class<?>[]{FactoryBean.class};
+    }
+
+    @Test
+    public void scheduledFiringUsesCustomExecutorsThreadFactory() throws Exception {
+        final ManagedScheduledExecutorService custom = InitialContext.doLookup(MSES_JNDI);
+
+        // Collect thread ids from the custom MSES's primary pool by submitting short-lived
+        // probes that each block until a release latch fires. Capturing several probes
+        // forces the primary pool to spawn each of its worker threads concurrently.
+        final int primarySamples = 3;
+        final Set<Long> primaryThreadIds = ConcurrentHashMap.newKeySet();
+        final CountDownLatch primaryStarted = new CountDownLatch(primarySamples);
+        final CountDownLatch release = new CountDownLatch(1);
+        final Set<Future<?>> primaryProbes = new HashSet<>();
+        for (int i = 0; i < primarySamples; i++) {
+            primaryProbes.add(custom.submit(() -> {
+                primaryThreadIds.add(Thread.currentThread().getId());
+                primaryStarted.countDown();
+                try {
+                    release.await(5, TimeUnit.SECONDS);
+                } catch (final InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                }
+            }));
+        }
+        assertTrue("Primary-pool probes should start", primaryStarted.await(5, TimeUnit.SECONDS));
+        release.countDown();
+        for (final Future<?> f : primaryProbes) {
+            f.get(5, TimeUnit.SECONDS);
+        }
+        assertFalse("Primary pool should have reported at least one worker thread id", primaryThreadIds.isEmpty());
+
+        final AtomicInteger counter = new AtomicInteger();
+        final CompletableFuture<ThreadSnapshot> future = bean.capture(counter);
+        final ThreadSnapshot firing = future.get(15, TimeUnit.SECONDS);
+
+        assertNotNull("Scheduled firing must complete", firing);
+        assertFalse("Firing must run on the secondary pool, not the primary — the custom executor's "
+                        + "primary core-size is capped by maxAsync (firing thread id=" + firing.threadId
+                        + ", primary ids=" + primaryThreadIds + ")",
+                primaryThreadIds.contains(firing.threadId));
+        assertTrue("Firing thread must come from the custom MSES's ManagedThreadFactory "
+                        + "(expected name prefix '" + ManagedThreadFactoryImpl.DEFAULT_PREFIX
+                        + "', got '" + firing.threadName + "')",
+                firing.threadName.startsWith(ManagedThreadFactoryImpl.DEFAULT_PREFIX));
+        assertTrue("Firing thread must be a ManageableThread produced by ManagedThreadFactoryImpl "
+                        + "(got class " + firing.threadClass + ")",
+                firing.manageable);
+    }
+
+    @ManagedScheduledExecutorDefinition(name = MSES_JNDI, maxAsync = 3)
+    @ApplicationScoped
+    public static class FactoryBean {
+
+        @Asynchronous(executor = MSES_JNDI, runAt = @Schedule(cron = "* * * * * *"))
+        public CompletableFuture<ThreadSnapshot> capture(final AtomicInteger counter) {
+            counter.incrementAndGet();
+            final Thread t = Thread.currentThread();
+            final ThreadSnapshot snap = new ThreadSnapshot(
+                    t.getId(),
+                    t.getName(),
+                    t.getClass().getName(),
+                    t instanceof ManageableThread);
+            final CompletableFuture<ThreadSnapshot> future = Asynchronous.Result.getFuture();
+            future.complete(snap);
+            return future;
+        }
+    }
+
+    public record ThreadSnapshot(long threadId, String threadName, String threadClass, boolean manageable) {
+    }
+
+    @jakarta.ejb.Singleton
+    public static class DummyEjb {
+    }
+}