Merge branch 'main' into chore/cloud-testing

Author: hsilan

.github/workflows/build.yml

@@ -58,4 +58,4 @@ jobs:
           cache: maven
 
       - name: Build and test
-        run: mvn -B -q install --file pom.xml
+        run: mvn -B install --file pom.xml

README.md

@@ -312,9 +312,11 @@ public class OrderProcessor extends DurableHandler<Order, OrderResult> {
 |--------|-------------|---------|
 | `withLambdaClient()` | Custom AWS Lambda client | Auto-configured Lambda client |
 | `withSerDes()` | Serializer for step results | Jackson with default settings |
-| `withExecutorService()` | Thread pool for async step execution | Cached daemon thread pool |
+| `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 an SDK-managed thread pool.
+
 ## Logging
 
 The SDK provides a `DurableLogger` via `ctx.getLogger()` that automatically includes execution metadata in log entries and suppresses duplicate logs during replay.

docs/design.md

@@ -86,9 +86,52 @@ public class MyHandler extends DurableHandler<Input, Output> {
 |--------|---------|
 | `lambdaClient` | Auto-created `LambdaClient` for current region, primed for performance (see [`DurableConfig.java`](../sdk/src/main/java/com/amazonaws/lambda/durable/DurableConfig.java)) |
 | `serDes` | `JacksonSerDes` |
-| `executor` | `Executors.newCachedThreadPool()` |
+| `executorService` | `Executors.newCachedThreadPool()` (for user-defined operations only) |
 | `loggerConfig` | `LoggerConfig.defaults()` (suppress replay logs) |
 
+### Thread Pool Architecture
+
+The SDK uses two separate thread pools with distinct responsibilities:
+
+**User Executor (`DurableConfig.executorService`):**
+- Runs user-defined operations (the code passed to `ctx.step()` and `ctx.stepAsync()`)
+- Configurable via `DurableConfig.builder().withExecutorService()`
+- Default: cached daemon thread pool
+
+**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** | 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
+@Override
+protected DurableConfig createConfiguration() {
+    var executor = new ThreadPoolExecutor(
+        4, 10,                          // core/max threads
+        60L, TimeUnit.SECONDS,          // idle timeout
+        new LinkedBlockingQueue<>(100), // bounded queue
+        new ThreadFactoryBuilder()
+            .setNameFormat("order-processor-%d")
+            .setDaemon(true)
+            .build());
+
+    return DurableConfig.builder()
+        .withExecutorService(executor)
+        .build();
+}
+```
+
 ### Step Configuration
 
 ```java

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

@@ -172,7 +172,10 @@ private static DurableExecutionClient createDefaultDurableExecutionClient() {
     }
 
     /**
-     * Creates a default ExecutorService for durable execution. Uses a cached thread pool with daemon threads.
+     * Creates a default ExecutorService for running user-defined operations. Uses a cached thread pool with daemon
+     * threads by default.
+     *
+     * <p>This executor is used exclusively for user operations. Internal SDK coordination uses the common ForkJoinPool.
      *
      * @return Default ExecutorService instance
      */
@@ -252,7 +255,11 @@ public Builder withSerDes(SerDes serDes) {
         }
 
         /**
-         * Sets a custom ExecutorService. If not set, a default cached thread pool will be created.
+         * Sets a custom ExecutorService for running user-defined operations. If not set, a default cached thread pool
+         * will be created.
+         *
+         * <p>This executor is used exclusively for running user-defined operations. Internal SDK coordination (polling,
+         * checkpointing) uses the common ForkJoinPool and is not affected by this setting.
          *
          * @param executorService Custom ExecutorService instance
          * @return This builder

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

@@ -6,13 +6,11 @@
 import com.amazonaws.lambda.durable.execution.ExecutionManager;
 import com.amazonaws.lambda.durable.execution.ThreadType;
 import com.amazonaws.lambda.durable.logging.DurableLogger;
-import com.amazonaws.lambda.durable.logging.LoggerConfig;
 import com.amazonaws.lambda.durable.operation.CallbackOperation;
 import com.amazonaws.lambda.durable.operation.InvokeOperation;
 import com.amazonaws.lambda.durable.operation.StepOperation;
 import com.amazonaws.lambda.durable.operation.WaitOperation;
 import com.amazonaws.lambda.durable.retry.RetryStrategies;
-import com.amazonaws.lambda.durable.serde.SerDes;
 import com.amazonaws.services.lambda.runtime.Context;
 import java.time.Duration;
 import java.util.Objects;
@@ -24,20 +22,16 @@
 
 public class DurableContext {
     private final ExecutionManager executionManager;
-    private final SerDes serDes;
+    private final DurableConfig durableConfig;
     private final Context lambdaContext;
     private final AtomicInteger operationCounter;
     private final DurableLogger logger;
     private final ExecutionContext executionContext;
 
     DurableContext(
-            ExecutionManager executionManager,
-            SerDes serDes,
-            Context lambdaContext,
-            LoggerConfig loggerConfig,
-            String contextId) {
+            ExecutionManager executionManager, DurableConfig durableConfig, Context lambdaContext, String contextId) {
         this.executionManager = executionManager;
-        this.serDes = serDes;
+        this.durableConfig = durableConfig;
         this.lambdaContext = lambdaContext;
         this.operationCounter = new AtomicInteger(0);
         this.executionContext = new ExecutionContext(executionManager.getDurableExecutionArn());
@@ -47,14 +41,14 @@ public class DurableContext {
                 LoggerFactory.getLogger(DurableContext.class),
                 executionManager,
                 requestId,
-                loggerConfig.suppressReplayLogs());
+                durableConfig.getLoggerConfig().suppressReplayLogs());
 
         // Register root context thread as active
         executionManager.registerActiveThreadWithContext(contextId, ThreadType.CONTEXT);
     }
 
-    DurableContext(ExecutionManager executionManager, SerDes serDes, Context lambdaContext, LoggerConfig loggerConfig) {
-        this(executionManager, serDes, lambdaContext, loggerConfig, "Root");
+    DurableContext(ExecutionManager executionManager, DurableConfig config, Context lambdaContext) {
+        this(executionManager, config, lambdaContext, "Root");
     }
 
     public <T> T step(String name, Class<T> resultType, Supplier<T> func) {
@@ -121,8 +115,16 @@ public <T> DurableFuture<T> stepAsync(String name, TypeToken<T> typeToken, Suppl
         }
 
         // Create and start step operation with TypeToken
-        StepOperation<T> operation =
-                new StepOperation<>(operationId, name, func, typeToken, config, executionManager, logger, serDes);
+        var operation = new StepOperation<>(
+                operationId,
+                name,
+                func,
+                typeToken,
+                config,
+                executionManager,
+                logger,
+                durableConfig.getSerDes(),
+                durableConfig.getExecutorService());
 
         operation.execute(); // Start the step (returns immediately)
 
@@ -209,7 +211,14 @@ public <T, U> DurableFuture<T> invokeAsync(
 
         // Create and start invoke operation
         var operation = new InvokeOperation<>(
-                operationId, name, functionName, payload, typeToken, config, executionManager, serDes);
+                operationId,
+                name,
+                functionName,
+                payload,
+                typeToken,
+                config,
+                executionManager,
+                durableConfig.getSerDes());
 
         operation.execute(); // checkpoint the invoke operation
         return new DurableFuture<>(operation); // Block (will throw SuspendExecutionException if needed)
@@ -246,7 +255,8 @@ public <T> DurableCallbackFuture<T> createCallback(String name, Class<T> resultT
             validateReplay(operationId, OperationType.CALLBACK, name, existing);
         }
 
-        var operation = new CallbackOperation<>(operationId, name, resultType, config, executionManager, serDes);
+        var operation = new CallbackOperation<>(
+                operationId, name, resultType, config, executionManager, durableConfig.getSerDes());
         operation.execute();
 
         return new DurableCallbackFuture<>(operation.getCallbackId(), operation);
@@ -264,7 +274,8 @@ public <T> DurableCallbackFuture<T> createCallback(String name, TypeToken<T> typ
             validateReplay(operationId, OperationType.CALLBACK, name, existing);
         }
 
-        var operation = new CallbackOperation<>(operationId, name, typeToken, config, executionManager, serDes);
+        var operation = new CallbackOperation<>(
+                operationId, name, typeToken, config, executionManager, durableConfig.getSerDes());
         operation.execute();
 
         return new DurableCallbackFuture<>(operation.getCallbackId(), operation);

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

@@ -52,16 +52,14 @@ public static <I, O> DurableExecutionOutput execute(
             throw new IllegalStateException("First operation must be EXECUTION");
         }
 
-        // Get executor from config (always non-null)
-        var executor = config.getExecutorService();
+        // Get executor from config for running user step functions
+        var userExecutor = config.getExecutorService();
 
-        // TODO: Should we pass the whole input instead?
         var executionManager = new ExecutionManager(
                 input.durableExecutionArn(),
                 input.checkpointToken(),
                 input.initialExecutionState(),
-                config.getDurableExecutionClient(),
-                executor);
+                config.getDurableExecutionClient());
 
         var executionOp = executionManager.getExecutionOperation();
         logger.debug("EXECUTION operation found: {}", executionOp.id());
@@ -74,11 +72,10 @@ public static <I, O> DurableExecutionOutput execute(
             var handlerFuture = CompletableFuture.supplyAsync(
                     () -> {
                         // Create context in the executor thread so it detects the correct thread name
-                        var context =
-                                new DurableContext(executionManager, serDes, lambdaContext, config.getLoggerConfig());
+                        var context = new DurableContext(executionManager, config, lambdaContext);
                         return handler.apply(userInput, context);
                     },
-                    executor);
+                    userExecutor);
 
             // Get suspend future from ExecutionManager. If this future completes, it
             // indicates
@@ -145,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,7 +7,6 @@
 import java.util.List;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutorService;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Supplier;
@@ -20,6 +19,11 @@
  *
  * <p>Single responsibility: Queue and batch checkpoint requests efficiently. Uses CheckpointCallback to notify when
  * checkpoints complete, avoiding cyclic dependency.
+ *
+ * <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
@@ -30,19 +34,16 @@ class CheckpointBatcher {
     private final String durableExecutionArn;
     private final DurableExecutionClient client;
     private final BlockingQueue<CheckpointRequest> queue = new LinkedBlockingQueue<>();
-    private final ExecutorService executor;
     private final AtomicBoolean isProcessing = new AtomicBoolean(false);
 
     record CheckpointRequest(OperationUpdate update, CompletableFuture<Void> completion) {}
 
     CheckpointBatcher(
             DurableExecutionClient client,
-            ExecutorService executor,
             String durableExecutionArn,
             Supplier<String> tokenSupplier,
             CheckpointCallback callback) {
         this.client = client;
-        this.executor = executor;
         this.durableExecutionArn = durableExecutionArn;
         this.tokenSupplier = tokenSupplier;
         this.callback = callback;
@@ -56,7 +57,7 @@ CompletableFuture<Void> checkpoint(OperationUpdate update) {
         queue.offer(new CheckpointRequest(update, future));
 
         if (isProcessing.compareAndSet(false, true)) {
-            executor.submit(this::processQueue);
+            InternalExecutor.INSTANCE.execute(this::processQueue);
         }
 
         return future;
@@ -107,7 +108,7 @@ private void processQueue() {
             isProcessing.set(false);
 
             if (!queue.isEmpty() && isProcessing.compareAndSet(false, true)) {
-                executor.submit(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.Executors;
 import java.util.concurrent.Phaser;
 import java.util.concurrent.atomic.AtomicReference;
 import org.slf4j.Logger;
@@ -35,7 +33,10 @@
  *   <li>Polling (for waits and retries)
  * </ul>
  *
- * <p>This is the single entry point for all execution coordination.
+ * <p>This is the single entry point for all execution coordination. Internal coordination (polling, checkpointing) uses
+ * a dedicated SDK thread pool, while user-defined operations run on a customer-configured executor.
+ *
+ * @see InternalExecutor
  */
 public class ExecutionManager {
 
@@ -54,9 +55,6 @@ public class ExecutionManager {
     private final Map<String, Phaser> openPhasers = Collections.synchronizedMap(new HashMap<>());
     private final CompletableFuture<Void> suspendExecutionFuture = new CompletableFuture<>();
 
-    // ===== Executors =====
-    private final Executor managedExecutor;
-
     // ===== Checkpoint Batching =====
     private final CheckpointBatcher checkpointBatcher;
     private final DurableExecutionClient client;
@@ -65,8 +63,7 @@ public ExecutionManager(
             String durableExecutionArn,
             String checkpointToken,
             InitialExecutionState initialExecutionState,
-            DurableExecutionClient client,
-            Executor executor) {
+            DurableExecutionClient client) {
         this.durableExecutionArn = durableExecutionArn;
         this.checkpointToken = checkpointToken;
         this.client = client;
@@ -77,13 +74,9 @@ public ExecutionManager(
         this.executionMode =
                 new AtomicReference<>(operations.size() > 1 ? ExecutionMode.REPLAY : ExecutionMode.EXECUTION);
 
-        this.managedExecutor = executor;
-
-        // Create checkpoint manager (package-private)
-        // Pass method references to avoid cyclic dependency
-        var checkpointExecutor = Executors.newSingleThreadExecutor();
+        // Create checkpoint manager using common pool for internal coordination
         this.checkpointBatcher = new CheckpointBatcher(
-                client, checkpointExecutor, durableExecutionArn, this::getCheckpointToken, this::onCheckpointComplete);
+                client, durableExecutionArn, this::getCheckpointToken, this::onCheckpointComplete);
     }
 
     private void loadAllOperations(InitialExecutionState initialExecutionState) {
@@ -260,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) {
-        managedExecutor.execute(() -> {
+        InternalExecutor.INSTANCE.execute(() -> {
             // Sleep until the start
             try {
                 var sleepDuration = Duration.between(Instant.now(), firstPollTime);
@@ -300,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) {
-        managedExecutor.execute(() -> {
+        InternalExecutor.INSTANCE.execute(() -> {
             // Sleep until first poll time
             try {
                 Duration sleepDuration = Duration.between(Instant.now(), firstPollTime);
@@ -342,10 +335,6 @@ public void pollUntilReady(
 
     // ===== Utilities =====
 
-    public Executor getManagedExecutor() {
-        return managedExecutor;
-    }
-
     public CompletableFuture<Void> getSuspendExecutionFuture() {
         return suspendExecutionFuture;
     }