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 9d79f7de069e15e2b394de2580eda9c3ad7716be Author: Richard Zowalla <[email protected]> AuthorDate: Wed Sep 9 20:47:09 2026 +0200 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. --- .../org/apache/openejb/threads/task/CUTask.java | 48 +++--- .../openejb/threads/ThreadContextCaptureTest.java | 29 ++++ .../threads/task/CUTaskFailingContextTest.java | 170 +++++++++++++++++++++ 3 files changed, 226 insertions(+), 21 deletions(-) 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 8f6039a59e..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 @@ -76,27 +76,33 @@ public abstract class CUTask<T> extends ManagedTaskListenerTask implements Compa // 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]; - for (int i = 0; i < CONTAINER_LISTENERS.length; i++) { - oldStates[i] = CONTAINER_LISTENERS[i].onStart(containerListenerStates[i]); - } - } else { - oldStates = null; - } + + // 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; @@ -106,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); } - invocationContext.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 index d517967931..a0341d5058 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 @@ -82,6 +82,11 @@ public class ThreadContextCaptureTest { facade.checkInvocationContextNotPropagated(); } + @Test + public void applyingAContextOnAnotherThreadLeavesItsClassLoaderInPlace() throws Exception { + facade.checkWorkerThreadClassLoader(); + } + @Test public void currentContextExecutorCapturesWhereItWasCreated() throws Exception { facade.checkCurrentContextExecutorCaptureTime(); @@ -173,6 +178,30 @@ public class ThreadContextCaptureTest { 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); 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"; + } + } + +}
