feat(map): Implement Map Operation

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

@@ -0,0 +1,39 @@
+// 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.List;
+import software.amazon.lambda.durable.DurableContext;
+import software.amazon.lambda.durable.DurableHandler;
+
+/**
+ * Example demonstrating the map operation with the Durable Execution SDK.
+ *
+ * <p>This handler processes a list of names concurrently using {@code map()}, where each item runs in its own child
+ * context with full checkpoint-and-replay support.
+ *
+ * <ol>
+ *   <li>Create a list of names from the input
+ *   <li>Map over each name concurrently, applying a greeting transformation via a durable step
+ *   <li>Collect and join the results
+ * </ol>
+ */
+public class SimpleMapExample extends DurableHandler<GreetingRequest, String> {
+
+    @Override
+    public String handleRequest(GreetingRequest input, DurableContext context) {
+        var name = input.getName();
+        context.getLogger().info("Starting map example for {}", name);
+
+        var names = List.of(name, name.toUpperCase(), name.toLowerCase());
+
+        // Map over each name concurrently — each iteration runs in its own child context
+        var result = context.map("greet-all", names, String.class, (ctx, item, index) -> {
+            return ctx.step("greet-" + index, String.class, stepCtx -> "Hello, " + item + "!");
+        });
+
+        context.getLogger().info("Map completed: allSucceeded={}, size={}", result.allSucceeded(), result.size());
+
+        return String.join(" | ", result.results());
+    }
+}

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

@@ -539,4 +539,13 @@ void testManyAsyncChildContextExample(int steps, long maxExecutionTime, long max
         assertTrue(minimalReplayTimeMs < maxReplayTime);
         assertTrue(minimalExecutionTimeMs < maxExecutionTime);
     }
+
+    @Test
+    void testSimpleMapExample() {
+        var runner = CloudDurableTestRunner.create(arn("simple-map-example"), GreetingRequest.class, String.class);
+        var result = runner.run(new GreetingRequest("Alice"));
+
+        assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
+        assertEquals("Hello, Alice! | Hello, ALICE! | Hello, alice!", result.getResult(String.class));
+    }
 }

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

@@ -0,0 +1,48 @@
+// 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 org.junit.jupiter.api.Test;
+import software.amazon.lambda.durable.model.ExecutionStatus;
+import software.amazon.lambda.durable.testing.LocalDurableTestRunner;
+
+class SimpleMapExampleTest {
+
+    @Test
+    void testSimpleMapExample() {
+        var handler = new SimpleMapExample();
+        var runner = LocalDurableTestRunner.create(GreetingRequest.class, handler);
+
+        var result = runner.runUntilComplete(new GreetingRequest("Alice"));
+
+        assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
+        assertEquals("Hello, Alice! | Hello, ALICE! | Hello, alice!", result.getResult(String.class));
+    }
+
+    @Test
+    void testWithDefaultName() {
+        var handler = new SimpleMapExample();
+        var runner = LocalDurableTestRunner.create(GreetingRequest.class, handler);
+
+        var result = runner.runUntilComplete(new GreetingRequest());
+
+        assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
+        assertEquals("Hello, World! | Hello, WORLD! | Hello, world!", result.getResult(String.class));
+    }
+
+    @Test
+    void testReplay() {
+        var handler = new SimpleMapExample();
+        var runner = LocalDurableTestRunner.create(GreetingRequest.class, handler);
+
+        var input = new GreetingRequest("Bob");
+        var result1 = runner.runUntilComplete(input);
+        assertEquals("Hello, Bob! | Hello, BOB! | Hello, bob!", result1.getResult(String.class));
+
+        // Replay — should use cached results
+        var result2 = runner.runUntilComplete(input);
+        assertEquals(result1.getResult(String.class), result2.getResult(String.class));
+    }
+}

examples/template.yaml

@@ -468,6 +468,31 @@ Resources:
       DockerContext: ../
       DockerTag: durable-examples
 
+  SimpleMapExampleFunction:
+    Type: AWS::Serverless::Function
+    Properties:
+      PackageType: Image
+      FunctionName: !Join
+        - ''
+        - - 'simple-map-example'
+          - !Ref FunctionNameSuffix
+      ImageConfig:
+        Command: ["software.amazon.lambda.durable.examples.SimpleMapExample::handleRequest"]
+      DurableConfig:
+        ExecutionTimeout: 300
+        RetentionPeriodInDays: 7
+      Policies:
+        - Statement:
+            - Effect: Allow
+              Action:
+                - lambda:CheckpointDurableExecutions
+                - lambda:GetDurableExecutionState
+              Resource: !Sub "arn:aws:lambda:${AWS::Region}:${AWS::AccountId}:function:simple-map-example${FunctionNameSuffix}"
+    Metadata:
+      Dockerfile: !Ref DockerFile
+      DockerContext: ../
+      DockerTag: durable-examples
+
 Outputs:
   NoopExampleFunction:
     Description: Noop Example Function ARN
@@ -613,3 +638,11 @@ Outputs:
     Description: Many Async Child Context Example Function Name
     Value: !Ref ManyAsyncChildContextExampleFunction
 
+  SimpleMapExampleFunction:
+    Description: Simple Map Example Function ARN
+    Value: !GetAtt SimpleMapExampleFunction.Arn
+
+  SimpleMapExampleFunctionName:
+    Description: Simple Map Example Function Name
+    Value: !Ref SimpleMapExampleFunction
+

sdk-integration-tests/src/test/java/software/amazon/lambda/durable/MapIntegrationTest.java

@@ -0,0 +1,52 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+package software.amazon.lambda.durable;
+
+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 MapIntegrationTest {
+
+    @Test
+    void testSimpleMap() {
+        var runner = LocalDurableTestRunner.create(String.class, (input, context) -> {
+            var items = List.of("a", "b", "c");
+            var result = context.map("process-items", items, String.class, (ctx, item, index) -> {
+                return item.toUpperCase();
+            });
+
+            assertTrue(result.allSucceeded());
+            assertEquals(3, result.size());
+            assertEquals("A", result.getResult(0));
+            assertEquals("B", result.getResult(1));
+            assertEquals("C", result.getResult(2));
+
+            return String.join(",", result.results());
+        });
+
+        var result = runner.runUntilComplete("test");
+        assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
+        assertEquals("A,B,C", result.getResult(String.class));
+    }
+
+    @Test
+    void testMapWithStepsInsideBranches() {
+        var runner = LocalDurableTestRunner.create(String.class, (input, context) -> {
+            var items = List.of("hello", "world");
+            var result = context.map("map-with-steps", items, String.class, (ctx, item, index) -> {
+                return ctx.step("process-" + index, String.class, stepCtx -> item.toUpperCase());
+            });
+
+            assertTrue(result.allSucceeded());
+            return String.join(" ", result.results());
+        });
+
+        var result = runner.runUntilComplete("test");
+        assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
+        assertEquals("HELLO WORLD", result.getResult(String.class));
+    }
+}

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

@@ -0,0 +1,23 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+package software.amazon.lambda.durable;
+
+/**
+ * A {@link DurableFuture} that is already completed with a value.
+ *
+ * <p>Used for short-circuit cases (e.g., empty collection in map) where no checkpoint or async execution is needed.
+ *
+ * @param <T> the result type
+ */
+class CompletedDurableFuture<T> implements DurableFuture<T> {
+    private final T value;
+
+    CompletedDurableFuture(T value) {
+        this.value = value;
+    }
+
+    @Override
+    public T get() {
+        return value;
+    }
+}

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

@@ -0,0 +1,51 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+package software.amazon.lambda.durable;
+
+/**
+ * Controls when a concurrent operation (map or parallel) completes.
+ *
+ * <p>Provides factory methods for common completion strategies and fine-grained control via {@code minSuccessful},
+ * {@code toleratedFailureCount}, and {@code toleratedFailurePercentage}.
+ */
+public class CompletionConfig {
+    private final Integer minSuccessful;
+    private final Integer toleratedFailureCount;
+    private final Double toleratedFailurePercentage;
+
+    private CompletionConfig(Integer minSuccessful, Integer toleratedFailureCount, Double toleratedFailurePercentage) {
+        this.minSuccessful = minSuccessful;
+        this.toleratedFailureCount = toleratedFailureCount;
+        this.toleratedFailurePercentage = toleratedFailurePercentage;
+    }
+
+    /** All items must succeed. Zero failures tolerated. */
+    public static CompletionConfig allSuccessful() {
+        return new CompletionConfig(null, 0, null);
+    }
+
+    /** All items run regardless of failures. Failures captured per-item. */
+    public static CompletionConfig allCompleted() {
+        return new CompletionConfig(null, null, null);
+    }
+
+    /** Complete as soon as the first item succeeds. */
+    public static CompletionConfig firstSuccessful() {
+        return new CompletionConfig(1, null, null);
+    }
+
+    /** @return minimum number of successful items required, or null if not set */
+    public Integer minSuccessful() {
+        return minSuccessful;
+    }
+
+    /** @return maximum number of failures tolerated, or null if unlimited */
+    public Integer toleratedFailureCount() {
+        return toleratedFailureCount;
+    }
+
+    /** @return maximum percentage of failures tolerated (0.0 to 1.0), or null if not set */
+    public Double toleratedFailurePercentage() {
+        return toleratedFailurePercentage;
+    }
+}

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

@@ -4,6 +4,8 @@
 
 import com.amazonaws.services.lambda.runtime.Context;
 import java.time.Duration;
+import java.util.Collection;
+import java.util.List;
 import java.util.Objects;
 import java.util.function.BiConsumer;
 import java.util.function.Function;
@@ -14,11 +16,13 @@
 import software.amazon.lambda.durable.execution.OperationIdGenerator;
 import software.amazon.lambda.durable.execution.ThreadType;
 import software.amazon.lambda.durable.logging.DurableLogger;
+import software.amazon.lambda.durable.model.BatchResult;
 import software.amazon.lambda.durable.model.OperationIdentifier;
 import software.amazon.lambda.durable.model.OperationSubType;
 import software.amazon.lambda.durable.operation.CallbackOperation;
 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.StepOperation;
 import software.amazon.lambda.durable.operation.WaitOperation;
 import software.amazon.lambda.durable.validation.ParameterValidator;
@@ -361,6 +365,83 @@ private <T> DurableFuture<T> runInChildContextAsync(
         return operation;
     }
 
+    // ========== map methods ==========
+
+    public <I, O> BatchResult<O> map(
+            String name, Collection<I> items, Class<O> resultType, MapFunction<I, O> function) {
+        return mapAsync(
+                        name,
+                        items,
+                        TypeToken.get(resultType),
+                        function,
+                        MapConfig.builder().build())
+                .get();
+    }
+
+    public <I, O> BatchResult<O> map(
+            String name, Collection<I> items, Class<O> resultType, MapFunction<I, O> function, MapConfig config) {
+        return mapAsync(name, items, TypeToken.get(resultType), function, config)
+                .get();
+    }
+
+    public <I, O> BatchResult<O> map(
+            String name, Collection<I> items, TypeToken<O> resultType, MapFunction<I, O> function) {
+        return mapAsync(name, items, resultType, function, MapConfig.builder().build())
+                .get();
+    }
+
+    public <I, O> BatchResult<O> map(
+            String name, Collection<I> items, TypeToken<O> resultType, MapFunction<I, O> function, MapConfig config) {
+        return mapAsync(name, items, resultType, function, config).get();
+    }
+
+    public <I, O> DurableFuture<BatchResult<O>> mapAsync(
+            String name, Collection<I> items, Class<O> resultType, MapFunction<I, O> function) {
+        return mapAsync(
+                name,
+                items,
+                TypeToken.get(resultType),
+                function,
+                MapConfig.builder().build());
+    }
+
+    public <I, O> DurableFuture<BatchResult<O>> mapAsync(
+            String name, Collection<I> items, Class<O> resultType, MapFunction<I, O> function, MapConfig config) {
+        return mapAsync(name, items, TypeToken.get(resultType), function, config);
+    }
+
+    public <I, O> DurableFuture<BatchResult<O>> mapAsync(
+            String name, Collection<I> items, TypeToken<O> resultType, MapFunction<I, O> function) {
+        return mapAsync(name, items, resultType, function, MapConfig.builder().build());
+    }
+
+    public <I, O> DurableFuture<BatchResult<O>> mapAsync(
+            String name, Collection<I> items, TypeToken<O> resultType, MapFunction<I, O> function, MapConfig config) {
+        Objects.requireNonNull(items, "items cannot be null");
+        Objects.requireNonNull(function, "function cannot be null");
+        Objects.requireNonNull(resultType, "resultType cannot be null");
+        Objects.requireNonNull(config, "config cannot be null");
+        ParameterValidator.validateOperationName(name);
+        ParameterValidator.validateOrderedCollection(items);
+
+        if (config.serDes() == null) {
+            config = config.toBuilder().serDes(getDurableConfig().getSerDes()).build();
+        }
+
+        // Short-circuit for empty collections — no checkpoint overhead
+        if (items.isEmpty()) {
+            return new CompletedDurableFuture<>(BatchResult.empty());
+        }
+
+        // Convert to List for deterministic index-based access
+        var itemList = List.copyOf(items);
+        var operationId = nextOperationId();
+
+        var operation = new MapOperation<>(operationId, name, itemList, function, resultType, config, this);
+        operation.execute();
+        return operation;
+    }
+
     // ========= waitForCallback methods =============
     public <T> T waitForCallback(String name, Class<T> resultType, BiConsumer<String, StepContext> func) {
         return waitForCallbackAsync(