feature: ConcurrencyOperation and Parallel (#201)

* Add parallel design
- Implement ConcurrencyOperation
- Implement ParallelOperation, ParallelContext
* Add parallel examples and unit tests
---------

Co-authored-by: Alex Wang <wangyb@amazon.com>

Author: wangyb-A

docs/core/parallel.md

@@ -0,0 +1,143 @@
+# Parallel Operations Design Plan
+
+## Overview
+
+Add parallel execution capability to the AWS Lambda Durable Execution SDK, allowing multiple branches to run concurrently within a single durable function execution.
+
+## API Design
+
+### User Interface
+
+```java
+try (var parallelContext = ctx.parallel(ParallelConfig.builder().build())) {
+    DurableFuture<Boolean> task1 = parallelContext.branch("validate", Boolean.class, branchContext -> validate());
+    DurableFuture<String> task2 = parallelContext.branch("process", String.class, branchContext -> process());
+    parallelContext.join(); // Wait for completion based on config
+    
+    // Access results
+    Boolean validated = task1.get();
+    String processed = task2.get();
+}
+```
+
+### Core Components
+
+#### 1. ParallelConfig
+Configuration object controlling parallel execution behavior:
+
+```java
+ParallelConfig config = ParallelConfig.builder()
+    .maxConcurrency(5)           // Max branches running simultaneously
+    .minSuccessful(3)            // Minimum successful branches required (-1 = all)
+    .toleratedFailureCount(2)    // Max failures before stopping execution
+    .build();
+```
+
+**Configuration Rules:**
+- `maxConcurrency`: Controls resource usage, prevents overwhelming the system
+- `minSuccessful`: Enables "best effort" scenarios where not all branches need to succeed
+- `toleratedFailureCount`: Fail-fast behavior when too many branches fail
+
+#### 2. ParallelContext
+Manages the lifecycle of parallel branches:
+
+```java
+public class ParallelContext implements AutoCloseable {
+    // Create branches
+    public <T> DurableFuture<T> branch(String name, Class<T> resultType, Function<DurableContext, T> func);
+    public <T> DurableFuture<T> branch(String name, TypeToken<T> resultType, Function<DurableContext, T> func);
+    
+    // Wait for completion
+    public void join();
+    
+    // AutoCloseable ensures join() is called
+    public void close();
+}
+```
+
+#### 3. DurableContext Integration
+Add single method to existing `DurableContext`:
+
+```java
+public ParallelContext parallel(ParallelConfig config);
+```
+
+## Implementation Strategy
+
+### 1. Leverage Existing Child Context Infrastructure
+
+Each parallel branch will be implemented as a `ChildContextOperation`:
+- **Isolation**: Each branch has its own checkpoint log
+- **Replay Safety**: Branches replay independently
+- **Error Handling**: Branch failures don't affect other branches directly
+
+### 2. Execution Flow
+
+1. **Branch Registration**: `branch()` calls create `ChildContextOperation` instances but don't execute immediately
+2. **Execution Start**: `join()` triggers execution of branches respecting `maxConcurrency`
+3. **Concurrency Control**: Use a queue to manage pending branches when `maxConcurrency` is reached
+4. **Completion Logic**: Monitor success/failure counts against configuration thresholds
+5. **Result Collection**: Return results via `DurableFuture` instances
+
+
+### 4. Error Handling Strategy
+
+**Branch-Level Failures:**
+- Individual branch failures are captured in their respective `DurableFuture`
+- Don't immediately fail the entire parallel operation
+- Count towards `failureCount` for threshold checking
+
+**Parallel-Level Failures:**
+- Exceed `toleratedFailureCount`: Stop starting new branches, wait for running ones
+- Insufficient `minSuccessful`: Throw `ParallelExecutionException` after all branches complete
+- Configuration validation errors: Fail immediately
+
+## Key Design Decisions
+
+### 1. Build on Child Contexts
+- **Pros**: Reuses existing isolation and checkpointing logic
+- **Cons**: Each branch has overhead of a separate child context
+- **Decision**: Acceptable trade-off for clean isolation and replay safety
+
+### 2. Eager vs Lazy Execution
+- **Chosen**: Lazy execution (branches start only on `join()`)
+- **Rationale**: Allows all branches to be registered before execution starts, enabling better concurrency planning
+
+### 3. AutoCloseable Pattern
+- **Purpose**: Ensures `join()` is called even if user forgets
+- **Behavior**: If `close()` is called before `join()`, automatically call `join()`
+
+### 4. Configuration Validation
+- Validate at `ParallelConfig.build()` time:
+  - `maxConcurrency > 0`
+  - `minSuccessful >= -1` (where -1 means "all")
+  - `toleratedFailureCount >= 0`
+  - `minSuccessful + toleratedFailureCount <= total branches` (validated at runtime)
+
+## Implementation Files
+
+### New Files to Create
+1. `ParallelConfig.java` - Configuration builder
+2. `ParallelContext.java` - User-facing parallel context
+3. `operation/ParallelOperation.java` - Core execution logic
+4. `exception/ParallelExecutionException.java` - Parallel-specific exceptions
+
+### Files to Modify
+1. `DurableContext.java` - Add `parallel()` method
+2. `DurableFuture.java` - Ensure compatibility with parallel results (likely no changes needed)
+
+## Testing Strategy
+
+### Unit Tests
+- `ParallelConfigTest` - Configuration validation
+- `ParallelOperationTest` - Core execution logic with mocked child contexts
+
+### Integration Tests
+- Success scenarios with various configurations
+- Failure scenarios (exceeding thresholds)
+- Concurrency limits
+- Replay behavior
+
+### Example Implementation
+- `ParallelExample.java` in examples module
+- Demonstrate common patterns and error handling

examples/src/main/java/software/amazon/lambda/durable/examples/ParallelExample.java

@@ -0,0 +1,58 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+package software.amazon.lambda.durable.examples;
+
+import java.util.ArrayList;
+import java.util.List;
+import software.amazon.lambda.durable.DurableContext;
+import software.amazon.lambda.durable.DurableFuture;
+import software.amazon.lambda.durable.DurableHandler;
+import software.amazon.lambda.durable.ParallelConfig;
+
+/**
+ * Example demonstrating parallel branch execution with the Durable Execution SDK.
+ *
+ * <p>This handler processes a list of items concurrently using {@code context.parallel()}:
+ *
+ * <ol>
+ *   <li>Each item is processed in its own branch (child context)
+ *   <li>All branches run concurrently and their results are collected
+ *   <li>A final step combines the results into a summary
+ * </ol>
+ *
+ * <p>The {@link software.amazon.lambda.durable.ParallelContext} implements {@link AutoCloseable}, so try-with-resources
+ * guarantees {@code join()} is called even if an exception occurs.
+ */
+public class ParallelExample extends DurableHandler<ParallelExample.Input, ParallelExample.Output> {
+
+    public record Input(List<String> items) {}
+
+    public record Output(List<String> results, int totalProcessed) {}
+
+    @Override
+    public Output handleRequest(Input input, DurableContext context) {
+        var logger = context.getLogger();
+        var items = input.items();
+        logger.info("Starting parallel processing of {} items", items.size());
+
+        var config = ParallelConfig.builder().build();
+
+        var futures = new ArrayList<DurableFuture<String>>(items.size());
+
+        try (var parallel = context.parallel("process-items", config)) {
+            for (var item : items) {
+                var future = parallel.branch("process-" + item, String.class, branchCtx -> {
+                    branchCtx.getLogger().info("Processing item: {}", item);
+                    return branchCtx.step("transform-" + item, String.class, stepCtx -> item.toUpperCase());
+                });
+                futures.add(future);
+            }
+        } // join() called here via AutoCloseable
+
+        logger.info("All branches complete, collecting results");
+
+        var results = futures.stream().map(DurableFuture::get).toList();
+
+        return new Output(results, results.size());
+    }
+}

examples/src/test/java/software/amazon/lambda/durable/examples/ParallelExampleTest.java

@@ -0,0 +1,60 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+package software.amazon.lambda.durable.examples;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import java.util.List;
+import org.junit.jupiter.api.Test;
+import software.amazon.lambda.durable.model.ExecutionStatus;
+import software.amazon.lambda.durable.testing.LocalDurableTestRunner;
+
+class ParallelExampleTest {
+
+    @Test
+    void testParallelExampleRunsSuccessfully() {
+        var handler = new ParallelExample();
+        var runner = LocalDurableTestRunner.create(ParallelExample.Input.class, handler);
+
+        var input = new ParallelExample.Input(List.of("apple", "banana", "cherry"));
+        var result = runner.runUntilComplete(input);
+
+        assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
+
+        var output = result.getResult(ParallelExample.Output.class);
+        assertEquals(3, output.totalProcessed());
+        assertTrue(output.results().contains("APPLE"));
+        assertTrue(output.results().contains("BANANA"));
+        assertTrue(output.results().contains("CHERRY"));
+    }
+
+    @Test
+    void testParallelExampleWithSingleItem() {
+        var handler = new ParallelExample();
+        var runner = LocalDurableTestRunner.create(ParallelExample.Input.class, handler);
+
+        var input = new ParallelExample.Input(List.of("hello"));
+        var result = runner.runUntilComplete(input);
+
+        assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
+
+        var output = result.getResult(ParallelExample.Output.class);
+        assertEquals(1, output.totalProcessed());
+        assertEquals(List.of("HELLO"), output.results());
+    }
+
+    @Test
+    void testParallelExampleWithEmptyInput() {
+        var handler = new ParallelExample();
+        var runner = LocalDurableTestRunner.create(ParallelExample.Input.class, handler);
+
+        var input = new ParallelExample.Input(List.of());
+        var result = runner.runUntilComplete(input);
+
+        assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
+
+        var output = result.getResult(ParallelExample.Output.class);
+        assertEquals(0, output.totalProcessed());
+        assertTrue(output.results().isEmpty());
+    }
+}

sdk/pom.xml

@@ -63,6 +63,12 @@
             <artifactId>mockito-core</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-simple</artifactId>
+            <version>${slf4j.version}</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>

sdk/src/main/java/software/amazon/lambda/durable/DurableContext.java

@@ -23,6 +23,7 @@
 import software.amazon.lambda.durable.operation.ChildContextOperation;
 import software.amazon.lambda.durable.operation.InvokeOperation;
 import software.amazon.lambda.durable.operation.MapOperation;
+import software.amazon.lambda.durable.operation.ParallelOperation;
 import software.amazon.lambda.durable.operation.StepOperation;
 import software.amazon.lambda.durable.operation.WaitOperation;
 import software.amazon.lambda.durable.validation.ParameterValidator;
@@ -596,6 +597,32 @@ public <I, O> DurableFuture<MapResult<O>> mapAsync(
         return operation;
     }
 
+    // ========== parallel methods ==========
+
+    /**
+     * Creates a {@link ParallelContext} for executing multiple branches concurrently.
+     *
+     * @param config the parallel execution configuration
+     * @return a new ParallelContext for registering and executing branches
+     */
+    public ParallelContext parallel(String name, ParallelConfig config) {
+        Objects.requireNonNull(config, "config cannot be null");
+        var operationId = nextOperationId();
+
+        var parallelOp = new ParallelOperation<>(
+                OperationIdentifier.of(operationId, name, OperationType.CONTEXT, OperationSubType.PARALLEL),
+                TypeToken.get(Void.class),
+                getDurableConfig().getSerDes(),
+                this,
+                config.maxConcurrency(),
+                config.minSuccessful(),
+                config.toleratedFailureCount());
+
+        parallelOp.execute();
+
+        return new ParallelContext(parallelOp, this);
+    }
+
     // ========= waitForCallback methods =============
 
     /**