This is an automated email from the ASF dual-hosted git repository. rzo1 pushed a commit to branch tomee-10.x in repository https://gitbox.apache.org/repos/asf/tomee.git
commit 8004a2aeffb38dc87b331b21c874e47cc47ee844 Author: Richard Zowalla <[email protected]> AuthorDate: Sun Sep 20 18:13:27 2026 +0200 TOMEE-4703 / TOMEE-4704 capture thread context on the submitting thread (#2940) * TOMEE-4703 restore the caller's class loader after exiting the ThreadContext ApplicationThreadContextRestorer.endContext() restored the class loader it had saved and then called ThreadContext.exit(), which sets the loader again from the value the context recorded on entry. That value is the application class loader installed by begin(), so the thread was left with the application class loader instead of the one it arrived with. Exit first and restore the loader afterwards. This affects pool threads, where the captured context is never the thread's current context and the identity guard in begin() therefore never applies. * TOMEE-4703 capture thread context eagerly on the submitting thread ApplicationThreadContextProvider stored a live reference to the submitting thread's ThreadContext and copied it later, in begin(), on the thread running the task. A ThreadContext is confined to its thread, so that copy raced with the owner and could fail with a ConcurrentModificationException (TOMEE-4699). Capture an immutable snapshot in currentContext() instead and build a new ThreadContext from it in begin(). The snapshot cannot hold a ThreadContext, because enter() modifies its argument and fails if that context was already entered, so one snapshot could not be applied twice. Context data that is tied to the invocation the capture is taken from is left out. InvocationContext is part of the interceptor chain the caller is still in, and BaseContext.getContextData() exposes its unsynchronized map to application code. DestroyContext references the captured context and would keep it reachable for the lifetime of the capture. Both are recreated on the thread the context is entered on. The context data is also formatted outside the map's monitor now, so that application hashCode() implementations no longer run under a lock that is taken on every invocation. * 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. * TOMEE-4704 report and clean up failures while establishing a task's context CUTask.invoke() entered the CUTask context, started the container listeners and applied the context snapshot before the try block. An exception from any of them skipped taskStarting, taskAborted and taskDone, so the ManagedTaskListener was never notified, and skipped the finally block, which left the CUTask context set on the thread. The next task to run on that thread then found a context it had not entered. Move the setup inside the try. The listener is notified and the teardown runs whether the failure comes from establishing the context or from the task itself. The teardown only unwinds what was established: the container listeners that were started, and the context only if it was entered. CONTAINER_LISTENERS is read once, since the array is replaced when a listener is registered and the teardown walks the listeners that were started. * TOMEE-4703 avoid double negation when filtering non-propagated context data (cherry picked from commit af5c06fb82a674f80fe660238a6dde28b7fe2d3c) --- .../org/apache/openejb/core/ThreadContext.java | 92 ++++++- .../impl/ApplicationThreadContextProvider.java | 32 ++- .../openejb/threads/impl/ContextServiceImpl.java | 6 +- .../apache/openejb/threads/task/CURunnable.java | 5 + .../org/apache/openejb/threads/task/CUTask.java | 67 +++-- .../openejb/threads/ThreadContextCaptureTest.java | 282 +++++++++++++++++++++ .../threads/task/CUTaskFailingContextTest.java | 170 +++++++++++++ 7 files changed, 613 insertions(+), 41 deletions(-) diff --git a/container/openejb-core/src/main/java/org/apache/openejb/core/ThreadContext.java b/container/openejb-core/src/main/java/org/apache/openejb/core/ThreadContext.java index c421e35084..a904a2f147 100644 --- a/container/openejb-core/src/main/java/org/apache/openejb/core/ThreadContext.java +++ b/container/openejb-core/src/main/java/org/apache/openejb/core/ThreadContext.java @@ -25,6 +25,7 @@ import org.apache.openejb.util.Logger; import java.util.Collections; import java.util.HashMap; import java.util.Map; +import java.util.Set; import java.util.concurrent.CopyOnWriteArraySet; import java.util.concurrent.atomic.AtomicBoolean; import java.util.stream.Collectors; @@ -151,6 +152,11 @@ public class ThreadContext { this.currentOperation = operation; } + /** + * Copy constructor. Must be called on the thread that owns <code>that</code>, since a + * ThreadContext is confined to its thread. Use {@link #capture()} to pass a context to + * another thread. + */ public ThreadContext(final ThreadContext that) { this.beanContext = that.beanContext; this.primaryKey = that.primaryKey; @@ -160,6 +166,85 @@ public class ThreadContext { this.oldClassLoader = that.oldClassLoader; } + /** + * Returns an immutable copy of the calling thread's context, which may be passed to other + * threads. Must be called on the thread that owns the context. + * + * @return the capture, or <code>null</code> if no context is entered on this thread + */ + public static Capture capture() { + final ThreadContext current = threadStorage.get(); + return current == null ? null : new Capture(current); + } + + /** + * Immutable copy of the state a {@link ThreadContext} propagates: bean context, primary key and + * context data. Per-thread state such as the class loader to restore, the entered flag and the + * current operation is not included. + * <p> + * A capture may be applied to any number of threads, including concurrently. + * {@link #newThreadContext()} returns a separate mutable {@link ThreadContext} for each caller, + * since {@link ThreadContext#enter(ThreadContext)} modifies its argument and fails if that + * context was already entered. + */ + public static final class Capture { + + /** + * Context data tied to the invocation a capture is taken from, listed by class name to avoid + * a dependency on the types. It is not propagated: + * <ul> + * <li><code>InvocationContext</code> is part of the interceptor chain the calling thread is + * still in. It is single use, and {@link BaseContext#getContextData()} exposes its + * unsynchronized map to application code.</li> + * <li><code>DestroyContext</code> references the captured context and would keep it + * reachable for the lifetime of the capture. A new one is created when the context is + * entered on another thread.</li> + * </ul> + */ + private static final Set<String> NON_PROPAGATED = Set.of( + "jakarta.interceptor.InvocationContext", + "org.apache.openejb.cdi.RequestScopedThreadContextListener$DestroyContext"); + + private final BeanContext beanContext; + private final Object primaryKey; + private final Map<Class, Object> data; + + private Capture(final ThreadContext that) { + this.beanContext = that.beanContext; + this.primaryKey = that.primaryKey; + + final Map<Class, Object> copy = new HashMap<>(); + synchronized (that.data) { + for (final Map.Entry<Class, Object> entry : that.data.entrySet()) { + if (NON_PROPAGATED.contains(entry.getKey().getName())) { + continue; + } + copy.put(entry.getKey(), entry.getValue()); + } + } + this.data = Collections.unmodifiableMap(copy); + } + + /** + * @return a new mutable {@link ThreadContext} with the captured state, for the calling thread + * to pass to {@link ThreadContext#enter(ThreadContext)} + */ + public ThreadContext newThreadContext() { + final ThreadContext context = new ThreadContext(beanContext, primaryKey); + context.data.putAll(data); + return context; + } + + @Override + public String toString() { + return "ThreadContext.Capture{" + + "beanContext=" + beanContext.getId() + + ", primaryKey=" + primaryKey + + ", " + dataToString(data) + + '}'; + } + } + public BeanContext getBeanContext() { return beanContext; } @@ -237,12 +322,15 @@ public class ThreadContext { '}'; } - private String dataToString(final Map<Class, Object> data) { + private static String dataToString(final Map<Class, Object> data) { + // iterating a synchronized map requires its monitor, see TOMEE-4699. Copy under the monitor + // and format outside of it, so that application hashCode() implementations do not run while + // a lock that is taken on every invocation is held. final Map<Class, Object> copy; - // copy data under monitor (synchronized map), format outside lock synchronized (data) { copy = new HashMap<>(data); } + return "data(" + copy.size() + ")=" + copy.entrySet().stream() .map(entry -> entry.getKey() + "=" + (entry.getValue() == null ? "null" : entry.getValue().hashCode())) .collect(Collectors.joining(", ")); diff --git a/container/openejb-core/src/main/java/org/apache/openejb/threads/impl/ApplicationThreadContextProvider.java b/container/openejb-core/src/main/java/org/apache/openejb/threads/impl/ApplicationThreadContextProvider.java index 5977812e3b..412b4f0ca0 100755 --- a/container/openejb-core/src/main/java/org/apache/openejb/threads/impl/ApplicationThreadContextProvider.java +++ b/container/openejb-core/src/main/java/org/apache/openejb/threads/impl/ApplicationThreadContextProvider.java @@ -39,7 +39,9 @@ public class ApplicationThreadContextProvider implements ThreadContextProvider, return clearedContext(props); } - return new ApplicationThreadContextSnapshot(appContext.getId(), ThreadContext.getThreadContext()); + // capture on the thread that owns the ThreadContext, see TOMEE-4699. A ThreadContext is + // confined to its thread; reading it from the thread running the task races with the owner. + return new ApplicationThreadContextSnapshot(appContext.getId(), ThreadContext.capture()); } @Override @@ -54,11 +56,11 @@ public class ApplicationThreadContextProvider implements ThreadContextProvider, public static class ApplicationThreadContextSnapshot implements ThreadContextSnapshot, Serializable { private final Object appId; - private final ThreadContext threadContext; + private final ThreadContext.Capture capturedThreadContext; - public ApplicationThreadContextSnapshot(final Object appId, final ThreadContext threadContext) { + public ApplicationThreadContextSnapshot(final Object appId, final ThreadContext.Capture capturedThreadContext) { this.appId = appId; - this.threadContext = threadContext; + this.capturedThreadContext = capturedThreadContext; } @Override @@ -71,9 +73,12 @@ public class ApplicationThreadContextProvider implements ThreadContextProvider, final ClassLoader oldCl = Thread.currentThread().getContextClassLoader(); Thread.currentThread().setContextClassLoader(appContext.getClassLoader()); - // Don't touch ThreadContext if it is already correct or none was captured - boolean changeThreadContext = threadContext != null && threadContext != ThreadContext.getThreadContext(); - ThreadContext oldThreadContext = changeThreadContext ? ThreadContext.enter(new ThreadContext(threadContext)) : null; + // leave the ThreadContext alone if nothing was captured, otherwise enter a new copy. This + // snapshot may be applied to any number of threads, including concurrently, and + // ThreadContext.enter modifies the context it is given. + final boolean changeThreadContext = capturedThreadContext != null; + final ThreadContext oldThreadContext = + changeThreadContext ? ThreadContext.enter(capturedThreadContext.newThreadContext()) : null; return new ApplicationThreadContextRestorer(oldCl, oldThreadContext, changeThreadContext); } @@ -81,7 +86,7 @@ public class ApplicationThreadContextProvider implements ThreadContextProvider, public String toString() { return "ApplicationThreadContextSnapshot@" + System.identityHashCode(this) + "{appId=" + appId + - "{threadContext=" + threadContext + + "{capturedThreadContext=" + capturedThreadContext + '}'; } @@ -100,13 +105,16 @@ public class ApplicationThreadContextProvider implements ThreadContextProvider, @Override public void endContext() throws IllegalStateException { - if (oldClassLoader != null) { - Thread.currentThread().setContextClassLoader(oldClassLoader); - } - + // exit before restoring the class loader. ThreadContext.exit sets the loader to the value + // the context recorded on entry, which is the application class loader installed by + // begin(), so restoring afterwards leaves the thread with the loader it started with. if (exitThreadContext) { ThreadContext.exit(oldThreadContext); } + + if (oldClassLoader != null) { + Thread.currentThread().setContextClassLoader(oldClassLoader); + } } @Override 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 5f41c025e4..696c49c3de 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; @@ -138,7 +139,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..a691342e8f 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,27 +73,36 @@ public abstract class CUTask<T> extends ManagedTaskListenerTask implements Compa } protected T invoke(final Callable<T> call) throws Exception { - initialContext.enter(); - final Object[] oldStates; - if (CONTAINER_LISTENERS.length > 0) { - oldStates = new Object[CONTAINER_LISTENERS.length]; - for (int i = 0; i < CONTAINER_LISTENERS.length; i++) { - oldStates[i] = CONTAINER_LISTENERS[i].onStart(containerListenerStates[i]); - } - } else { - oldStates = null; - } + // 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(); + + // read once: the array is replaced when a listener is registered, and the teardown below + // walks the listeners that were started + final ContainerListener[] listeners = CONTAINER_LISTENERS; + final Object[] oldStates = listeners.length > 0 ? new Object[listeners.length] : null; + int started = 0; ContextServiceImpl.State state = null; + boolean entered = false; + Throwable throwable = null; - if (contextService != null && snapshot != null) { - state = contextService.enter(snapshot); - } + // establishing the context can fail, see TOMEE-4699. Keep it inside the try so that the task + // listener is notified and the thread is cleaned up in that case as well. + try { + invocationContext.enter(); + entered = true; + for (int i = 0; i < listeners.length; i++) { + oldStates[i] = listeners[i].onStart(containerListenerStates[i]); + started = i + 1; + } - Throwable throwable = null; - try { - taskStarting(future, executor, delegate); // do it in try to avoid issues if an exception is thrown + if (contextService != null && snapshot != null) { + state = contextService.enter(snapshot); + } + + taskStarting(future, executor, delegate); return call.call(); } catch (final Throwable t) { throwable = t; @@ -97,15 +112,15 @@ public abstract class CUTask<T> extends ManagedTaskListenerTask implements Compa try { taskDone(future, executor, delegate, throwable); } finally { - if (CONTAINER_LISTENERS.length > 0) { - for (int i = 0; i < CONTAINER_LISTENERS.length; i++) { - CONTAINER_LISTENERS[i].onEnd(oldStates[i]); - } + for (int i = 0; i < started; i++) { + listeners[i].onEnd(oldStates[i]); } if (contextService != null && state != null) { contextService.exit(state); } - initialContext.exit(); + if (entered) { + 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 new file mode 100644 index 0000000000..a0341d5058 --- /dev/null +++ b/container/openejb-core/src/test/java/org/apache/openejb/threads/ThreadContextCaptureTest.java @@ -0,0 +1,282 @@ +/** + * 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.threads; + +import org.apache.openejb.core.ThreadContext; +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.ContextServiceImpl; +import org.junit.Test; +import org.junit.runner.RunWith; + +import jakarta.annotation.Resource; +import jakarta.ejb.EJB; +import jakarta.ejb.Singleton; +import jakarta.enterprise.concurrent.ContextService; +import jakarta.enterprise.concurrent.ManagedExecutorService; +import jakarta.interceptor.InvocationContext; +import java.net.URL; +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; +import java.util.concurrent.TimeUnit; + +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +/** + * TOMEE-4699: the submitting thread's ThreadContext is captured by value, on that thread, rather than + * read later from the thread running the task. + */ +@RunWith(ApplicationComposer.class) +public class ThreadContextCaptureTest { + @Module + public EnterpriseBean bean() { + return new SingletonBean(CaptureFacade.class).localBean(); + } + + @EJB + private CaptureFacade facade; + + @Test + public void contextIsCapturedWhenTheContextualProxyIsCreated() throws Exception { + facade.checkCaptureTime(); + } + + @Test + public void submittingWhileMutatingTheCallerContextDoesNotFail() throws Exception { + facade.submitWhileMutating(); + } + + @Test + public void inlineExecutionLeavesTheCallersClassLoaderInPlace() throws Exception { + facade.checkInlineClassLoader(); + } + + @Test + public void theCallersInvocationContextIsNotPropagated() throws Exception { + facade.checkInvocationContextNotPropagated(); + } + + @Test + public void applyingAContextOnAnotherThreadLeavesItsClassLoaderInPlace() throws Exception { + facade.checkWorkerThreadClassLoader(); + } + + @Test + public void currentContextExecutorCapturesWhereItWasCreated() throws Exception { + facade.checkCurrentContextExecutorCaptureTime(); + } + + @Test + public void aContextualProxyCanRunOnSeveralThreadsAtOnce() throws Exception { + facade.checkContextualProxyIsReusable(); + } + + public static class BeforeCapture { + } + + public static class AfterCapture { + } + + public static class Churn { + } + + @Singleton + public static class CaptureFacade { + @Resource + private ContextService contextService; + + @Resource + private ManagedExecutorService executorService; + + public void checkCaptureTime() throws Exception { + final ThreadContext caller = ThreadContext.getThreadContext(); + assertNotNull(caller); + + caller.set(BeforeCapture.class, new BeforeCapture()); + + final Callable<Object[]> contextual = contextService.contextualCallable(() -> { + final ThreadContext taskContext = ThreadContext.getThreadContext(); + assertNotNull(taskContext); + return new Object[]{taskContext.get(BeforeCapture.class), taskContext.get(AfterCapture.class)}; + }); + + // the caller keeps updating its own thread confined ThreadContext after the capture + caller.set(AfterCapture.class, new AfterCapture()); + + final ExecutorService plain = Executors.newSingleThreadExecutor(); + try { + final Object[] seen = plain.submit(contextual).get(1, TimeUnit.MINUTES); + assertNotNull("state present at capture time must be propagated", seen[0]); + assertNull("state added after the capture must not leak into the task", seen[1]); + } finally { + plain.shutdownNow(); + caller.remove(BeforeCapture.class); + caller.remove(AfterCapture.class); + } + } + + public void checkInlineClassLoader() { + final Thread thread = Thread.currentThread(); + final ClassLoader original = thread.getContextClassLoader(); + + // take the snapshot under the real thread context class loader, since the application is + // resolved from it and any other loader would produce a cleared snapshot + final ContextServiceImpl impl = ContextServiceImpl.class.cast(contextService); + final ContextServiceImpl.Snapshot snapshot = impl.snapshot(null); + + // a loader that is neither the application's nor the bean's, so that a restore to either + // of those is visible here + final ClassLoader marker = new URLClassLoader(new URL[0], original); + thread.setContextClassLoader(marker); + try { + impl.exit(impl.enter(snapshot)); + + assertSame("applying and restoring a context leaves the thread's loader unchanged", + marker, thread.getContextClassLoader()); + } finally { + thread.setContextClassLoader(original); + } + } + + public void checkInvocationContextNotPropagated() throws Exception { + final ThreadContext caller = ThreadContext.getThreadContext(); + assertNotNull(caller); + // set by the interceptor stack on the way into this method + assertNotNull("precondition: the caller is inside an invocation", + caller.get(InvocationContext.class)); + + final Future<InvocationContext> seen = executorService.submit( + () -> ThreadContext.getThreadContext().get(InvocationContext.class)); + + assertNull("the caller's InvocationContext must not be propagated to the task", + seen.get(1, TimeUnit.MINUTES)); + } + + public void checkWorkerThreadClassLoader() throws Exception { + final ContextServiceImpl impl = ContextServiceImpl.class.cast(contextService); + final ContextServiceImpl.Snapshot snapshot = impl.snapshot(null); + + // a pool thread applies a context and must be left with its own loader afterwards, + // not with the application's + final ExecutorService worker = Executors.newSingleThreadExecutor(); + try { + worker.submit(() -> { + final Thread thread = Thread.currentThread(); + final ClassLoader marker = new URLClassLoader(new URL[0], thread.getContextClassLoader()); + thread.setContextClassLoader(marker); + + impl.exit(impl.enter(snapshot)); + + assertSame("a worker thread keeps its own context class loader", + marker, thread.getContextClassLoader()); + return null; + }).get(1, TimeUnit.MINUTES); + } finally { + worker.shutdownNow(); + } + } + + 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); + + final List<Future<Boolean>> futures = new ArrayList<>(); + for (int i = 0; i < 200; i++) { + futures.add(executorService.submit(() -> ThreadContext.getThreadContext() != null)); + + // the interceptor stack updates the caller's context like this while the tasks start + for (int j = 0; j < 200; j++) { + caller.set(Churn.class, new Churn()); + caller.remove(Churn.class); + } + } + + for (final Future<Boolean> future : futures) { + assertTrue(future.get(1, TimeUnit.MINUTES)); + } + } + } +} diff --git a/container/openejb-core/src/test/java/org/apache/openejb/threads/task/CUTaskFailingContextTest.java b/container/openejb-core/src/test/java/org/apache/openejb/threads/task/CUTaskFailingContextTest.java new file mode 100644 index 0000000000..7e30e9c4ae --- /dev/null +++ b/container/openejb-core/src/test/java/org/apache/openejb/threads/task/CUTaskFailingContextTest.java @@ -0,0 +1,170 @@ +/** + * 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.threads.task; + +import org.apache.openejb.threads.impl.ContextServiceImpl; +import org.junit.Test; + +import jakarta.enterprise.concurrent.ManagedExecutorService; +import jakarta.enterprise.concurrent.ManagedTask; +import jakarta.enterprise.concurrent.ManagedTaskListener; +import jakarta.enterprise.concurrent.spi.ThreadContextProvider; +import jakarta.enterprise.concurrent.spi.ThreadContextRestorer; +import jakarta.enterprise.concurrent.spi.ThreadContextSnapshot; +import java.util.Map; +import java.util.concurrent.Callable; +import java.util.concurrent.Future; + +import static java.util.Collections.emptyList; +import static java.util.Collections.singletonList; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.fail; + +/** + * TOMEE-4704: when establishing the context fails, the task listener is still notified and the + * thread is left as it was found. + */ +public class CUTaskFailingContextTest { + @Test + public void aFailureEstablishingTheContextIsReportedAndCleanedUp() { + final IllegalStateException failure = new IllegalStateException("begin() blew up"); + final RecordingListener listener = new RecordingListener(); + + final CUCallable<String> task = new CUCallable<>( + new ListenedCallable(listener, () -> "never reached"), + new ContextServiceImpl(singletonList(new FailingProvider(failure)), emptyList(), emptyList())); + + try { + task.call(); + fail("the failure establishing the context must be propagated"); + } catch (final Exception e) { + assertSame(failure, rootCause(e)); + } + + assertSame("the listener is notified that the task was aborted", failure, rootCause(listener.aborted)); + assertSame("and that it is done", failure, rootCause(listener.done)); + assertFalse("the task never ran", listener.started); + + // the thread must be usable for the next task. A context left behind here is what turned a + // single failure into a pool thread that failed every task after it. + assertNull(CUTask.Context.CURRENT.get()); + } + + @Test + public void aTaskThatRunsNormallyStillReportsInOrder() throws Exception { + final RecordingListener listener = new RecordingListener(); + + final CUCallable<String> task = new CUCallable<>( + new ListenedCallable(listener, () -> "done"), + new ContextServiceImpl(emptyList(), emptyList(), emptyList())); + + assertEquals("done", task.call()); + assertNull(listener.aborted); + assertNull(listener.done); + assertNull(CUTask.Context.CURRENT.get()); + } + + private static Throwable rootCause(final Throwable t) { + Throwable current = t; + while (current.getCause() != null) { + current = current.getCause(); + } + return current; + } + + private static final class RecordingListener implements ManagedTaskListener { + private boolean started; + private Throwable aborted; + private Throwable done; + + @Override + public void taskSubmitted(final Future<?> future, final ManagedExecutorService executor, final Object task) { + } + + @Override + public void taskStarting(final Future<?> future, final ManagedExecutorService executor, final Object task) { + started = true; + } + + @Override + public void taskAborted(final Future<?> future, final ManagedExecutorService executor, final Object task, + final Throwable exception) { + aborted = exception; + } + + @Override + public void taskDone(final Future<?> future, final ManagedExecutorService executor, final Object task, + final Throwable exception) { + done = exception; + } + } + + private static final class ListenedCallable implements Callable<String>, ManagedTask { + private final ManagedTaskListener listener; + private final Callable<String> delegate; + + private ListenedCallable(final ManagedTaskListener listener, final Callable<String> delegate) { + this.listener = listener; + this.delegate = delegate; + } + + @Override + public ManagedTaskListener getManagedTaskListener() { + return listener; + } + + @Override + public Map<String, String> getExecutionProperties() { + return null; + } + + @Override + public String call() throws Exception { + return delegate.call(); + } + } + + private static final class FailingProvider implements ThreadContextProvider { + private final RuntimeException failure; + + private FailingProvider(final RuntimeException failure) { + this.failure = failure; + } + + @Override + public ThreadContextSnapshot currentContext(final Map<String, String> props) { + return () -> { + throw failure; + }; + } + + @Override + public ThreadContextSnapshot clearedContext(final Map<String, String> props) { + return () -> (ThreadContextRestorer) () -> { + }; + } + + @Override + public String getThreadContextType() { + return "failing-for-test"; + } + } + +}
