This is an automated email from the ASF dual-hosted git repository.
rzo1 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/tomee.git
The following commit(s) were added to refs/heads/main by this push:
new af5c06fb82 TOMEE-4703 / TOMEE-4704 capture thread context on the
submitting thread (#2940)
af5c06fb82 is described below
commit af5c06fb82a674f80fe660238a6dde28b7fe2d3c
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
---
.../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 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..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";
+ }
+ }
+
+}
