This is an automated email from the ASF dual-hosted git repository. rzo1 pushed a commit to branch TOMEE-4703 in repository https://gitbox.apache.org/repos/asf/tomee.git
commit c6cb5080c9817973022bc851a2e12ee907ccc5e6 Author: Richard Zowalla <[email protected]> AuthorDate: Wed Sep 9 20:47:09 2026 +0200 TOMEE-4703 fix context capture in currentContextExecutor and per-task CUTask.Context currentContextExecutor() was "command -> contextualRunnable(command).run()", so the context was captured inside execute(), on the thread submitting a command. ContextService specifies an executor that runs tasks on the calling thread "but with context that is captured from the thread that invokes currentContextExecutor". Capture the snapshot when the executor is created and reuse it, through a CUTask constructor that takes an existing snapshot. CUTask created one CUTask.Context per task, in the constructor, but a contextual proxy runs its task more than once and may run it on several threads at the same time. Concurrent invocations shared that Context, raced on its previous field and its exit task list, and failed with "Can't enter a context twice" when the callers had a context of their own. Create the Context per invocation. --- .../openejb/threads/impl/ContextServiceImpl.java | 6 +- .../apache/openejb/threads/task/CURunnable.java | 5 ++ .../org/apache/openejb/threads/task/CUTask.java | 23 +++++--- .../openejb/threads/ThreadContextCaptureTest.java | 69 ++++++++++++++++++++++ 4 files changed, 95 insertions(+), 8 deletions(-) diff --git a/container/openejb-core/src/main/java/org/apache/openejb/threads/impl/ContextServiceImpl.java b/container/openejb-core/src/main/java/org/apache/openejb/threads/impl/ContextServiceImpl.java index a10dd44016..6e7c51e4a8 100644 --- a/container/openejb-core/src/main/java/org/apache/openejb/threads/impl/ContextServiceImpl.java +++ b/container/openejb-core/src/main/java/org/apache/openejb/threads/impl/ContextServiceImpl.java @@ -26,6 +26,7 @@ import jakarta.enterprise.concurrent.spi.ThreadContextSnapshot; import org.apache.openejb.OpenEJBRuntimeException; import org.apache.openejb.resource.thread.ManagedExecutorServiceImplFactory; import org.apache.openejb.threads.future.CUCompletableFuture; +import org.apache.openejb.threads.task.CURunnable; import org.apache.openejb.threads.task.CUTask; import org.apache.openejb.util.LogCategory; import org.apache.openejb.util.Logger; @@ -149,7 +150,10 @@ public class ContextServiceImpl implements ContextService, Serializable { @Override public Executor currentContextExecutor() { - return command -> contextualRunnable(command).run(); + // ContextService specifies "context that is captured from the thread that invokes + // currentContextExecutor", so capture here rather than in execute() + final Snapshot snapshot = snapshot(null); + return command -> new CURunnable(command, this, snapshot).run(); } @Override diff --git a/container/openejb-core/src/main/java/org/apache/openejb/threads/task/CURunnable.java b/container/openejb-core/src/main/java/org/apache/openejb/threads/task/CURunnable.java index a47b1bc836..ed04883b13 100644 --- a/container/openejb-core/src/main/java/org/apache/openejb/threads/task/CURunnable.java +++ b/container/openejb-core/src/main/java/org/apache/openejb/threads/task/CURunnable.java @@ -28,6 +28,11 @@ public class CURunnable extends CUTask<Void> implements Runnable { super(task, ContextServiceImplFactory.newPropagateEverythingContextService()); delegate = task; } + public CURunnable(final Runnable task, final ContextServiceImpl contextService, final ContextServiceImpl.Snapshot snapshot) { + super(task, contextService, snapshot); + this.delegate = task; + } + public CURunnable(final Runnable task, final ContextServiceImpl contextService) { super(task, contextService); delegate = task; diff --git a/container/openejb-core/src/main/java/org/apache/openejb/threads/task/CUTask.java b/container/openejb-core/src/main/java/org/apache/openejb/threads/task/CUTask.java index c73813a262..8f6039a59e 100644 --- a/container/openejb-core/src/main/java/org/apache/openejb/threads/task/CUTask.java +++ b/container/openejb-core/src/main/java/org/apache/openejb/threads/task/CUTask.java @@ -44,18 +44,24 @@ public abstract class CUTask<T> extends ManagedTaskListenerTask implements Compa protected final ContextServiceImpl contextService; private final ContextServiceImpl.Snapshot snapshot; private final Object[] containerListenerStates; - private final Context initialContext; public CUTask(final Object task, final ContextServiceImpl contextService) { - this(task, contextService, null); + this(task, contextService, (Map<String, String>) null); } public CUTask(final Object task, final ContextServiceImpl contextService, Map<String, String> props) { + this(task, contextService, contextService.snapshot(props)); + } + + /** + * Uses a snapshot captured earlier, on the thread the context is taken from. + * {@link ContextService#currentContextExecutor()} captures when the executor is created rather + * than when a task is submitted to it. + */ + public CUTask(final Object task, final ContextServiceImpl contextService, final ContextServiceImpl.Snapshot snapshot) { super(task); this.contextService = contextService; - - snapshot = contextService.snapshot(props); - initialContext = new Context(); + this.snapshot = snapshot; if (CONTAINER_LISTENERS.length > 0) { containerListenerStates = new Object[CONTAINER_LISTENERS.length]; for (int i = 0; i < CONTAINER_LISTENERS.length; i++) { @@ -67,7 +73,10 @@ public abstract class CUTask<T> extends ManagedTaskListenerTask implements Compa } protected T invoke(final Callable<T> call) throws Exception { - initialContext.enter(); + // one per invocation rather than one per task: a contextual proxy runs its task more than + // once, and may run it on several threads at the same time + final Context invocationContext = new Context(); + invocationContext.enter(); final Object[] oldStates; if (CONTAINER_LISTENERS.length > 0) { oldStates = new Object[CONTAINER_LISTENERS.length]; @@ -105,7 +114,7 @@ public abstract class CUTask<T> extends ManagedTaskListenerTask implements Compa if (contextService != null && state != null) { contextService.exit(state); } - initialContext.exit(); + invocationContext.exit(); } } } diff --git a/container/openejb-core/src/test/java/org/apache/openejb/threads/ThreadContextCaptureTest.java b/container/openejb-core/src/test/java/org/apache/openejb/threads/ThreadContextCaptureTest.java index 049b397855..d517967931 100644 --- a/container/openejb-core/src/test/java/org/apache/openejb/threads/ThreadContextCaptureTest.java +++ b/container/openejb-core/src/test/java/org/apache/openejb/threads/ThreadContextCaptureTest.java @@ -36,6 +36,8 @@ import java.net.URLClassLoader; import java.util.ArrayList; import java.util.List; import java.util.concurrent.Callable; +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.Executor; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; @@ -80,6 +82,16 @@ public class ThreadContextCaptureTest { facade.checkInvocationContextNotPropagated(); } + @Test + public void currentContextExecutorCapturesWhereItWasCreated() throws Exception { + facade.checkCurrentContextExecutorCaptureTime(); + } + + @Test + public void aContextualProxyCanRunOnSeveralThreadsAtOnce() throws Exception { + facade.checkContextualProxyIsReusable(); + } + public static class BeforeCapture { } @@ -161,6 +173,63 @@ public class ThreadContextCaptureTest { seen.get(1, TimeUnit.MINUTES)); } + public void checkCurrentContextExecutorCaptureTime() throws Exception { + final ThreadContext caller = ThreadContext.getThreadContext(); + assertNotNull(caller); + + caller.set(BeforeCapture.class, new BeforeCapture()); + final Executor executor = contextService.currentContextExecutor(); + caller.set(AfterCapture.class, new AfterCapture()); + + try { + final Object[] seen = new Object[2]; + executor.execute(() -> { + final ThreadContext taskContext = ThreadContext.getThreadContext(); + seen[0] = taskContext.get(BeforeCapture.class); + seen[1] = taskContext.get(AfterCapture.class); + }); + + assertNotNull("state present when the executor was created must be propagated", seen[0]); + assertNull("state added afterwards must not be, since the capture happens in" + + " currentContextExecutor() and not in execute()", seen[1]); + } finally { + caller.remove(BeforeCapture.class); + caller.remove(AfterCapture.class); + } + } + + public void checkContextualProxyIsReusable() throws Exception { + final int threads = 2; + final int rounds = 200; + final Callable<Boolean> contextual = contextService.contextualCallable( + () -> ThreadContext.getThreadContext() != null); + + // one proxy, called by two threads at the same time. The callers are managed tasks, so + // each already has a CUTask context of its own, which is required to observe a task + // scoped CUTask.Context. The SPI requires a snapshot to be applicable "to any number of + // threads, including concurrently". + final CyclicBarrier barrier = new CyclicBarrier(threads); + final List<Future<Boolean>> futures = new ArrayList<>(); + for (int i = 0; i < threads; i++) { + futures.add(executorService.submit(() -> { + for (int round = 0; round < rounds; round++) { + barrier.await(1, TimeUnit.MINUTES); + if (!contextual.call()) { + return false; + } + } + return true; + })); + } + + for (final Future<Boolean> future : futures) { + assertTrue(future.get(1, TimeUnit.MINUTES)); + } + + // and once more afterwards, sequentially + assertTrue(contextual.call()); + } + public void submitWhileMutating() throws Exception { final ThreadContext caller = ThreadContext.getThreadContext(); assertNotNull(caller);
