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 c940d7304267cdfb3ecc30919059972fe0e077d2 Author: Richard Zowalla <[email protected]> AuthorDate: Wed Sep 9 20:47:09 2026 +0200 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. --- .../org/apache/openejb/core/ThreadContext.java | 104 +++++++++++- .../impl/ApplicationThreadContextProvider.java | 21 ++- .../openejb/threads/ThreadContextCaptureTest.java | 184 +++++++++++++++++++++ 3 files changed, 295 insertions(+), 14 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 5ef55eb8a8..0d45aba895 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,13 +152,98 @@ 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; - this.data.putAll(that.data); + synchronized (that.data) { + this.data.putAll(that.data); + } 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())) { + 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 + + ", data=" + dataToString(data) + + '}'; + } + } + public BeanContext getBeanContext() { return beanContext; } @@ -226,8 +312,7 @@ public class ThreadContext { return "ThreadContext{" + "beanContext=" + beanContext.getId() + ", primaryKey=" + primaryKey + - ", data(" + data.size() + - ")=" + dataToString(data) + + ", data=" + dataToString(data) + ", oldClassLoader=" + oldClassLoader + ", currentOperation=" + currentOperation + ", invokedInterface=" + invokedInterface + @@ -236,10 +321,17 @@ public class ThreadContext { '}'; } - private String dataToString(final Map<Class, Object> data) { - return data.entrySet().stream() + 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; + synchronized (data) { + copy = new HashMap<>(data); + } + + return "(" + 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 ac925f58da..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 + '}'; } 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..049b397855 --- /dev/null +++ b/container/openejb-core/src/test/java/org/apache/openejb/threads/ThreadContextCaptureTest.java @@ -0,0 +1,184 @@ +/** + * 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.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(); + } + + 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 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)); + } + } + } +}
