Merge remote-tracking branch 'origin/main' into invoke

Author: zhongkechen

.gitignore

@@ -21,6 +21,7 @@ buildNumber.properties
 # Kiro and SOP
 .kiro
 .sop
+.factorypath
 
 # Local testing
 

AGENTS.md

@@ -94,6 +94,10 @@ public DurableExecutor(DurableExecutionClient client, SerDes serDes) {
 
 Prefer descriptive domain names: `model`, `execution`, `operation`, `serde`, `exception`
 
+### Logging in Examples
+
+Use `context.getLogger()` instead of SLF4J's `LoggerFactory` in example handlers. It includes execution metadata and suppresses duplicate logs during replay.
+
 ## Do Not
 
 - Add new dependencies without explicit approval

examples/README.md

@@ -87,6 +87,7 @@ mvn test -Dtest=CloudBasedIntegrationTest \
 | [WaitAtLeastExample](src/main/java/com/amazonaws/lambda/durable/examples/WaitAtLeastExample.java) | Concurrent `stepAsync()` with `wait()` |
 | [RetryInProcessExample](src/main/java/com/amazonaws/lambda/durable/examples/RetryInProcessExample.java) | In-process retry with concurrent operations |
 | [WaitAtLeastInProcessExample](src/main/java/com/amazonaws/lambda/durable/examples/WaitAtLeastInProcessExample.java) | Wait completes before async step (no suspension) |
+| [ManyAsyncStepsExample](src/main/java/com/amazonaws/lambda/durable/examples/ManyAsyncStepsExample.java) | Performance test with 500 concurrent async steps |
 
 ## Cleanup
 

examples/src/main/java/com/amazonaws/lambda/durable/examples/ManyAsyncStepsExample.java

@@ -0,0 +1,60 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+package com.amazonaws.lambda.durable.examples;
+
+import com.amazonaws.lambda.durable.DurableContext;
+import com.amazonaws.lambda.durable.DurableFuture;
+import com.amazonaws.lambda.durable.DurableHandler;
+import java.time.Duration;
+import java.util.ArrayList;
+
+/**
+ * Performance test example demonstrating concurrent async steps.
+ *
+ * <p>This example tests the SDK's ability to handle many concurrent operations:
+ *
+ * <ul>
+ *   <li>Creates async steps in a loop
+ *   <li>Each step performs a simple computation
+ *   <li>All results are collected and summed
+ * </ul>
+ */
+public class ManyAsyncStepsExample extends DurableHandler<ManyAsyncStepsExample.Input, String> {
+
+    private static final int STEP_COUNT = 500;
+
+    public record Input(int multiplier) {}
+
+    @Override
+    public String handleRequest(Input input, DurableContext context) {
+        var startTime = System.currentTimeMillis();
+        var multiplier = input.multiplier() > 0 ? input.multiplier() : 1;
+
+        context.getLogger().info("Starting {} async steps with multiplier {}", STEP_COUNT, multiplier);
+
+        // Create 100 async steps
+        var futures = new ArrayList<DurableFuture<Integer>>(STEP_COUNT);
+        for (var i = 0; i < STEP_COUNT; i++) {
+            var index = i;
+            var future = context.stepAsync("compute-" + i, Integer.class, () -> index * multiplier);
+            futures.add(future);
+        }
+
+        context.getLogger().info("All {} async steps created, collecting results", STEP_COUNT);
+
+        // Collect all results
+        var totalSum = 0L;
+        for (var future : futures) {
+            totalSum += future.get();
+        }
+
+        var executionTimeMs = System.currentTimeMillis() - startTime;
+        context.getLogger()
+                .info("Completed {} steps, total sum: {}, execution time: {}ms", STEP_COUNT, totalSum, executionTimeMs);
+
+        // Wait 10 seconds to test replay
+        context.wait("post-compute-wait", Duration.ofSeconds(10));
+
+        return String.format("Completed %d async steps. Sum: %d, Time: %dms", STEP_COUNT, totalSum, executionTimeMs);
+    }
+}

examples/src/test/java/com/amazonaws/lambda/durable/examples/CloudBasedIntegrationTest.java

@@ -332,4 +332,22 @@ void testCallbackExampleWithTimeout() {
         assertNotNull(approvalOp);
         assertEquals(OperationStatus.TIMED_OUT, approvalOp.getStatus());
     }
+
+    @Test
+    void testManyAsyncStepsExample() {
+        var runner = CloudDurableTestRunner.create(
+                arn("many-async-steps-example"), ManyAsyncStepsExample.Input.class, String.class);
+        var result = runner.run(new ManyAsyncStepsExample.Input(2));
+
+        assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
+
+        var finalResult = result.getResult(String.class);
+        assertNotNull(finalResult);
+        assertTrue(finalResult.contains("500 async steps"));
+        assertTrue(finalResult.contains("249500")); // Sum of 0..499 * 2
+
+        // Verify some operations are tracked
+        assertNotNull(runner.getOperation("compute-0"));
+        assertNotNull(runner.getOperation("compute-499"));
+    }
 }

examples/src/test/java/com/amazonaws/lambda/durable/examples/ManyAsyncStepsExampleTest.java

@@ -0,0 +1,59 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+package com.amazonaws.lambda.durable.examples;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import com.amazonaws.lambda.durable.model.ExecutionStatus;
+import com.amazonaws.lambda.durable.testing.LocalDurableTestRunner;
+import org.junit.jupiter.api.Test;
+
+class ManyAsyncStepsExampleTest {
+
+    @Test
+    void testManyAsyncSteps() {
+        var handler = new ManyAsyncStepsExample();
+        var runner = LocalDurableTestRunner.create(ManyAsyncStepsExample.Input.class, handler);
+
+        var input = new ManyAsyncStepsExample.Input(2);
+        var result = runner.runUntilComplete(input);
+
+        assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
+
+        var output = result.getResult(String.class);
+        assertNotNull(output);
+        assertTrue(output.contains("500 async steps"));
+
+        // Sum of 0..499 * 2 = 499 * 500 / 2 * 2 = 249500
+        assertTrue(output.contains("249500"));
+    }
+
+    @Test
+    void testManyAsyncStepsWithDefaultMultiplier() {
+        var handler = new ManyAsyncStepsExample();
+        var runner = LocalDurableTestRunner.create(ManyAsyncStepsExample.Input.class, handler);
+
+        var input = new ManyAsyncStepsExample.Input(1);
+        var result = runner.runUntilComplete(input);
+
+        assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
+
+        // Sum of 0..499 = 499 * 500 / 2 = 124750
+        assertTrue(result.getResult(String.class).contains("124750"));
+    }
+
+    @Test
+    void testOperationsAreTracked() {
+        var handler = new ManyAsyncStepsExample();
+        var runner = LocalDurableTestRunner.create(ManyAsyncStepsExample.Input.class, handler);
+
+        var result = runner.runUntilComplete(new ManyAsyncStepsExample.Input(1));
+
+        assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
+
+        // Verify some operations are tracked
+        assertNotNull(result.getOperation("compute-0"));
+        assertNotNull(result.getOperation("compute-499"));
+        assertNotNull(result.getOperation("compute-250"));
+    }
+}

examples/template.yaml

@@ -279,6 +279,28 @@ Resources:
       DockerContext: ../
       DockerTag: durable-examples
 
+  ManyAsyncStepsExampleFunction:
+    Type: AWS::Serverless::Function
+    Properties:
+      PackageType: Image
+      FunctionName: many-async-steps-example
+      ImageConfig:
+        Command: ["com.amazonaws.lambda.durable.examples.ManyAsyncStepsExample::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:many-async-steps-example"
+    Metadata:
+      Dockerfile: examples/Dockerfile
+      DockerContext: ../
+      DockerTag: durable-examples
+
 Outputs:
   SimpleStepExampleFunction:
     Description: Simple Step Example Function ARN
@@ -375,3 +397,11 @@ Outputs:
   CallbackExampleFunctionName:
     Description: Callback Example Function Name
     Value: !Ref CallbackExampleFunction
+
+  ManyAsyncStepsExampleFunction:
+    Description: Many Async Steps Example Function ARN
+    Value: !GetAtt ManyAsyncStepsExampleFunction.Arn
+
+  ManyAsyncStepsExampleFunctionName:
+    Description: Many Async Steps Example Function Name
+    Value: !Ref ManyAsyncStepsExampleFunction

sdk-testing/src/main/java/com/amazonaws/lambda/durable/testing/LocalDurableTestRunner.java

@@ -261,7 +261,8 @@ private DurableExecutionInput createDurableInput(I input) {
                 .build();
 
         // Load previous operations and include them in InitialExecutionState
-        var existingOps = storage.getExecutionState("arn:aws:lambda:us-east-1:123456789012:function:test", null)
+        var existingOps = storage.getExecutionState(
+                        "arn:aws:lambda:us-east-1:123456789012:function:test", "test-token", null)
                 .operations();
         var allOps = new ArrayList<>(List.of(executionOp));
         allOps.addAll(existingOps);

sdk-testing/src/main/java/com/amazonaws/lambda/durable/testing/LocalMemoryExecutionClient.java

@@ -42,7 +42,7 @@ public CheckpointDurableExecutionResponse checkpoint(String arn, String token, L
     }
 
     @Override
-    public GetDurableExecutionStateResponse getExecutionState(String arn, String marker) {
+    public GetDurableExecutionStateResponse getExecutionState(String arn, String checkpointToken, String marker) {
         return GetDurableExecutionStateResponse.builder()
                 .operations(operations.values())
                 .build();

sdk/src/main/java/com/amazonaws/lambda/durable/client/DurableExecutionClient.java

@@ -10,5 +10,5 @@
 public interface DurableExecutionClient {
     CheckpointDurableExecutionResponse checkpoint(String arn, String token, List<OperationUpdate> updates);
 
-    GetDurableExecutionStateResponse getExecutionState(String arn, String marker);
+    GetDurableExecutionStateResponse getExecutionState(String arn, String checkpointToken, String marker);
 }