Refactor to InternalExecutor with cached thread pool.

Author: phipag

README.md

@@ -315,7 +315,7 @@ public class OrderProcessor extends DurableHandler<Order, OrderResult> {
 | `withExecutorService()` | Thread pool for user-defined operations | Cached daemon thread pool |
 | `withLoggerConfig()` | Logger behavior configuration | Suppress logs during replay |
 
-The `withExecutorService()` option configures the thread pool used for running user-defined operations. Internal SDK coordination (checkpoint batching, polling) runs on the JVM's common `ForkJoinPool`.
+The `withExecutorService()` option configures the thread pool used for running user-defined operations. Internal SDK coordination (checkpoint batching, polling) runs on an SDK-managed thread pool.
 
 ## Logging
 

docs/design.md

@@ -98,19 +98,20 @@ The SDK uses two separate thread pools with distinct responsibilities:
 - Configurable via `DurableConfig.builder().withExecutorService()`
 - Default: cached daemon thread pool
 
-**Internal Executor (`ForkJoinPool.commonPool()`):**
+**Internal Executor (`InternalExecutor.INSTANCE`):**
 - Runs SDK coordination tasks: checkpoint batching, polling for wait completion, phaser management
+- Dedicated cached thread pool with daemon threads named `durable-sdk-internal-*`
 - Not configurable by users
 
 **Benefits of this separation:**
 
 | Benefit | Description |
 |---------|-------------|
 | **Isolation** | User operations can't starve SDK internals, and vice versa |
-| **No shutdown management** | The common pool is JVM-managed; SDK coordination continues even if the user's executor is shut down |
-| **Efficient resource usage** | Common pool uses work-stealing and scales with available processors |
-| **Daemon threads** | Common pool threads won't prevent JVM shutdown |
-| **Single configuration point** | Changing `INTERNAL_EXECUTOR` in one place affects all SDK coordination |
+| **No shutdown management** | Internal pool uses daemon threads; SDK coordination continues even if the user's executor is shut down |
+| **Efficient resource usage** | Cached thread pool creates threads on demand and reuses idle threads (60s timeout) |
+| **Daemon threads** | Internal threads won't prevent JVM shutdown |
+| **Single configuration point** | Changing `InternalExecutor.INSTANCE` in one place affects all SDK coordination |
 
 **Example: Custom thread pool for user operations:**
 ```java

sdk/src/main/java/com/amazonaws/lambda/durable/DurableExecutor.java

@@ -142,7 +142,7 @@ public static <I, O> DurableExecutionOutput execute(
 
             // We DO NOT shutdown the executor since it should stay warm for re-invokes against a warm Lambda runtime.
             // For example, a re-invoke after a wait should re-use the same executor instance from DurableConfig.
-            // executor.shutdown();
+            // userExecutor.shutdown();
         }
     }
 

sdk/src/main/java/com/amazonaws/lambda/durable/execution/CheckpointBatcher.java

@@ -7,8 +7,6 @@
 import java.util.List;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.Executor;
-import java.util.concurrent.ForkJoinPool;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Supplier;
@@ -22,16 +20,15 @@
  * <p>Single responsibility: Queue and batch checkpoint requests efficiently. Uses CheckpointCallback to notify when
  * checkpoints complete, avoiding cyclic dependency.
  *
- * <p>Uses the common ForkJoinPool for internal coordination, keeping checkpoint processing separate from
- * customer-configured executors.
+ * <p>Uses a dedicated SDK thread pool for internal coordination, keeping checkpoint processing separate from
+ * customer-configured executors used for user-defined operations.
+ *
+ * @see InternalExecutor
  */
 class CheckpointBatcher {
     private static final int MAX_BATCH_SIZE_BYTES = 750 * 1024; // 750KB
     private static final Logger logger = LoggerFactory.getLogger(CheckpointBatcher.class);
 
-    /** Internal executor for SDK coordination tasks (checkpoint queue processing). */
-    private static final Executor INTERNAL_EXECUTOR = ForkJoinPool.commonPool();
-
     private final CheckpointCallback callback;
     private final Supplier<String> tokenSupplier;
     private final String durableExecutionArn;
@@ -60,7 +57,7 @@ CompletableFuture<Void> checkpoint(OperationUpdate update) {
         queue.offer(new CheckpointRequest(update, future));
 
         if (isProcessing.compareAndSet(false, true)) {
-            INTERNAL_EXECUTOR.execute(this::processQueue);
+            InternalExecutor.INSTANCE.execute(this::processQueue);
         }
 
         return future;
@@ -111,7 +108,7 @@ private void processQueue() {
             isProcessing.set(false);
 
             if (!queue.isEmpty() && isProcessing.compareAndSet(false, true)) {
-                INTERNAL_EXECUTOR.execute(this::processQueue);
+                InternalExecutor.INSTANCE.execute(this::processQueue);
             }
         }
     }

sdk/src/main/java/com/amazonaws/lambda/durable/execution/ExecutionManager.java

@@ -12,8 +12,6 @@
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.Executor;
-import java.util.concurrent.ForkJoinPool;
 import java.util.concurrent.Phaser;
 import java.util.concurrent.atomic.AtomicReference;
 import org.slf4j.Logger;
@@ -36,15 +34,14 @@
  * </ul>
  *
  * <p>This is the single entry point for all execution coordination. Internal coordination (polling, checkpointing) uses
- * the common ForkJoinPool, while user operations run on a customer-configured executor.
+ * a dedicated SDK thread pool, while user-defined operations run on a customer-configured executor.
+ *
+ * @see InternalExecutor
  */
 public class ExecutionManager {
 
     private static final Logger logger = LoggerFactory.getLogger(ExecutionManager.class);
 
-    /** Internal executor for SDK coordination tasks (polling, phaser management). */
-    private static final Executor INTERNAL_EXECUTOR = ForkJoinPool.commonPool();
-
     // ===== Execution State =====
     private final Map<String, Operation> operations = new ConcurrentHashMap<>();
     private final String executionOperationId;
@@ -256,7 +253,7 @@ public CompletableFuture<Void> sendOperationUpdate(OperationUpdate update) {
     // wait while another thread is still running and we therefore are not
     // re-invoked because we never suspended.
     public void pollForOperationUpdates(String operationId, Instant firstPollTime, Duration period) {
-        INTERNAL_EXECUTOR.execute(() -> {
+        InternalExecutor.INSTANCE.execute(() -> {
             // Sleep until the start
             try {
                 var sleepDuration = Duration.between(Instant.now(), firstPollTime);
@@ -296,7 +293,7 @@ public void pollForOperationUpdates(String operationId, Instant firstPollTime, D
     // re-invoked because we never suspended.
     public void pollUntilReady(
             String operationId, CompletableFuture<Void> future, Instant firstPollTime, Duration period) {
-        INTERNAL_EXECUTOR.execute(() -> {
+        InternalExecutor.INSTANCE.execute(() -> {
             // Sleep until first poll time
             try {
                 Duration sleepDuration = Duration.between(Instant.now(), firstPollTime);

sdk/src/main/java/com/amazonaws/lambda/durable/execution/InternalExecutor.java

@@ -0,0 +1,35 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+package com.amazonaws.lambda.durable.execution;
+
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Shared executor for internal SDK coordination tasks.
+ *
+ * <p>This executor is used for SDK-internal operations such as checkpoint batching, polling for wait completion, and
+ * phaser coordination. It is separate from the user-configured executor (via {@code DurableConfig}) which runs
+ * user-defined operations.
+ *
+ * <p>Using a dedicated thread pool ensures SDK coordination tasks are isolated from user code.
+ */
+final class InternalExecutor {
+
+    private static final AtomicInteger THREAD_COUNTER = new AtomicInteger(0);
+
+    /**
+     * Shared executor for all SDK-internal coordination tasks. Uses a cached thread pool that creates threads on
+     * demand, reuses idle threads, and terminates threads after 60 seconds of inactivity by default.
+     */
+    static final Executor INSTANCE = Executors.newCachedThreadPool(runnable -> {
+        var thread = new Thread(runnable, "durable-sdk-internal-" + THREAD_COUNTER.getAndIncrement());
+        thread.setDaemon(true);
+        return thread;
+    });
+
+    private InternalExecutor() {
+        // Utility class
+    }
+}