Fix ContextClassLoader propagation for poll tasks (#2808)

When PollerBehaviorAutoscaling is enabled, poll requests complete
asynchronously on the common ForkJoinPool. Subsequent tasks executed
by the PollTaskExecutor inherit the ContextClassLoader from these
ForkJoinPool threads (AppClassLoader) rather than the thread that
started the WorkerFactory (e.g., Spring Boot's LaunchedClassLoader).

This fix captures the original ContextClassLoader within
ExecutorThreadFactory and PollTaskExecutor, explicitly setting it
on all newly created worker threads and virtual threads.

Fixes #2795

Co-authored-by: tconley1428 <tconley1428@gmail.com>
Co-authored-by: Spencer Judge <spencer@temporal.io>

Author: 3 people

temporal-sdk/src/main/java/io/temporal/internal/worker/ExecutorThreadFactory.java

@@ -7,18 +7,23 @@ class ExecutorThreadFactory implements ThreadFactory {
   private final String threadPrefix;
 
   private final Thread.UncaughtExceptionHandler uncaughtExceptionHandler;
+  private final ClassLoader contextClassLoader;
   private final AtomicInteger threadIndex = new AtomicInteger();
 
   public ExecutorThreadFactory(String threadPrefix, Thread.UncaughtExceptionHandler eh) {
     this.threadPrefix = threadPrefix;
     this.uncaughtExceptionHandler = eh;
+    this.contextClassLoader = Thread.currentThread().getContextClassLoader();
   }
 
   @Override
   public Thread newThread(Runnable r) {
     Thread result = new Thread(r);
     result.setName(threadPrefix + ": " + threadIndex.incrementAndGet());
     result.setUncaughtExceptionHandler(uncaughtExceptionHandler);
+    if (contextClassLoader != null) {
+      result.setContextClassLoader(contextClassLoader);
+    }
     return result;
   }
 }

temporal-sdk/src/main/java/io/temporal/internal/worker/PollTaskExecutor.java

@@ -47,11 +47,15 @@ public interface TaskHandler<TT> {
     } else if (useVirtualThreads) {
       // If virtual threads are enabled, we use a virtual thread executor.
       AtomicInteger threadIndex = new AtomicInteger();
+      ClassLoader contextClassLoader = Thread.currentThread().getContextClassLoader();
       this.taskExecutor =
           VirtualThreadDelegate.newVirtualThreadExecutor(
               (t) -> {
                 t.setName(this.pollThreadNamePrefix + ": " + threadIndex.incrementAndGet());
                 t.setUncaughtExceptionHandler(pollerOptions.getUncaughtExceptionHandler());
+                if (contextClassLoader != null) {
+                  t.setContextClassLoader(contextClassLoader);
+                }
               });
     } else {
       ThreadPoolExecutor threadPoolTaskExecutor =

temporal-sdk/src/test/java/io/temporal/internal/worker/PollTaskExecutorTest.java

@@ -0,0 +1,65 @@
+package io.temporal.internal.worker;
+
+import static org.junit.Assert.assertEquals;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import org.junit.Test;
+
+public class PollTaskExecutorTest {
+
+  @Test
+  public void testContextClassLoaderInherited() throws Exception {
+    PollerOptions pollerOptions =
+        PollerOptions.newBuilder().setPollThreadNamePrefix("test").build();
+
+    ClassLoader originalClassLoader = Thread.currentThread().getContextClassLoader();
+    ClassLoader testClassLoader = new ClassLoader() {};
+    Thread.currentThread().setContextClassLoader(testClassLoader);
+
+    AtomicReference<ClassLoader> executedClassLoader = new AtomicReference<>();
+
+    try {
+      PollTaskExecutor<String> executor =
+          new PollTaskExecutor<>(
+              "namespace",
+              "taskQueue",
+              "identity",
+              new PollTaskExecutor.TaskHandler<String>() {
+                @Override
+                public void handle(String task) {
+                  executedClassLoader.set(Thread.currentThread().getContextClassLoader());
+                }
+
+                @Override
+                public Throwable wrapFailure(String task, Throwable failure) {
+                  return failure;
+                }
+              },
+              pollerOptions,
+              1,
+              false);
+
+      // Execute on a different thread with a different context class loader to simulate
+      // ForkJoinPool
+      CompletableFuture<Void> future = new CompletableFuture<>();
+      Thread triggerThread =
+          new Thread(
+              () -> {
+                Thread.currentThread().setContextClassLoader(null);
+                executor.process("task");
+                future.complete(null);
+              });
+      triggerThread.start();
+      future.get(5, TimeUnit.SECONDS);
+
+      // Wait for task completion
+      executor.shutdown(new ShutdownManager(), false).get(5, TimeUnit.SECONDS);
+
+      assertEquals(testClassLoader, executedClassLoader.get());
+    } finally {
+      Thread.currentThread().setContextClassLoader(originalClassLoader);
+    }
+  }
+}