Implemented Cloud based testing for callback

Author: maschnetwork

AGENTS.md

@@ -64,7 +64,11 @@ import static org.junit.jupiter.api.Assertions.*;           // Tests
 import static java.util.Collections.emptyList;              // Factory methods
 import static com.amazonaws.lambda.durable.model.Status.*;  // Enums
 
-// AVOID fully qualified names in code
+// ALWAYS use proper imports, NEVER use fully qualified class names in code
+// Bad:  var lambda = software.amazon.awssdk.services.lambda.LambdaClient.create();
+// Good: import software.amazon.awssdk.services.lambda.LambdaClient;
+//       var lambda = LambdaClient.create();
+
 // Bad:  com.amazonaws.lambda.durable.model.Status.SUCCESS
 // Good: import static and use SUCCESS directly
 

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

@@ -36,10 +36,9 @@ public String handleRequest(ApprovalRequest input, DurableContext context) {
         });
 
         // Step 2: Create callback for external approval
-        // Configure with 24-hour timeout and 1-hour heartbeat timeout
+        // Configure with 5min timeout
         var config = CallbackConfig.builder()
-                .timeout(Duration.ofHours(24))
-                .heartbeatTimeout(Duration.ofHours(1))
+                .timeout(Duration.ofMinutes(5))
                 .build();
 
         var callback = context.createCallback("approval", String.class, config);

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

@@ -223,4 +223,41 @@ void testCustomConfigExample() {
         assertTrue(stepResult.contains("user_age"));
         assertTrue(stepResult.contains("email_address"));
     }
+
+    @Test
+    void testCallbackExample() throws Exception {
+        var runner = CloudDurableTestRunner.create(
+                arn("callback-example"), ApprovalRequest.class, String.class);
+
+        // Start async execution
+        var execution = runner.startAsync(new ApprovalRequest("Purchase order", 5000.0));
+
+        // Wait for callback to appear
+        execution.pollUntil(exec -> exec.hasCallback("approval"));
+
+        // Get callback ID
+        var callbackId = execution.getCallbackId("approval");
+        assertNotNull(callbackId);
+
+        // Complete the callback using AWS SDK
+        var lambda = software.amazon.awssdk.services.lambda.LambdaClient.create();
+        lambda.sendDurableExecutionCallbackSuccess(req -> req
+                .callbackId(callbackId)
+                .result(software.amazon.awssdk.core.SdkBytes.fromUtf8String("\"approved\"")));
+
+        // Wait for execution to complete
+        var result = execution.pollUntilComplete();
+        assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
+
+        var finalResult = result.getResult(String.class);
+        assertNotNull(finalResult);
+        assertTrue(finalResult.contains("Approval request for: Purchase order"));
+        assertTrue(finalResult.contains("5000"));
+        assertTrue(finalResult.contains("approved"));
+
+        // Verify all operations completed
+        assertNotNull(execution.getOperation("prepare"));
+        assertNotNull(execution.getOperation("log-callback-command"));
+        assertNotNull(execution.getOperation("process-approval"));
+    }
 }

examples/template.yaml

@@ -208,6 +208,28 @@ Resources:
       DockerContext: ../
       DockerTag: durable-examples
 
+  CallbackExampleFunction:
+    Type: AWS::Serverless::Function
+    Properties:
+      PackageType: Image
+      FunctionName: callback-example
+      ImageConfig:
+        Command: ["com.amazonaws.lambda.durable.examples.CallbackExample::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:callback-example"
+    Metadata:
+      Dockerfile: examples/Dockerfile
+      DockerContext: ../
+      DockerTag: durable-examples
+
 Outputs:
   SimpleStepExampleFunction:
     Description: Simple Step Example Function ARN
@@ -280,3 +302,11 @@ Outputs:
   LoggingExampleFunctionName:
     Description: Logging Example Function Name
     Value: !Ref LoggingExampleFunction
+
+  CallbackExampleFunction:
+    Description: Callback Example Function ARN
+    Value: !GetAtt CallbackExampleFunction.Arn
+
+  CallbackExampleFunctionName:
+    Description: Callback Example Function Name
+    Value: !Ref CallbackExampleFunction

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

@@ -0,0 +1,201 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package com.amazonaws.lambda.durable.testing;
+
+import com.amazonaws.lambda.durable.model.ExecutionStatus;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.List;
+import java.util.concurrent.TimeoutException;
+import java.util.function.Predicate;
+import software.amazon.awssdk.services.lambda.LambdaClient;
+import software.amazon.awssdk.services.lambda.model.Event;
+import software.amazon.awssdk.services.lambda.model.GetDurableExecutionHistoryRequest;
+import software.amazon.awssdk.services.lambda.model.ResourceNotFoundException;
+
+/**
+ * Handle for an asynchronously executing durable function.
+ * Allows incremental polling and inspection of execution state.
+ */
+public class AsyncExecution<O> {
+    private final String executionArn;
+    private final LambdaClient lambdaClient;
+    private final Class<O> outputType;
+    private final Duration pollInterval;
+    private final Duration timeout;
+    private final HistoryEventProcessor processor;
+    private List<Event> currentHistory;
+    private TestResult<O> currentResult;
+
+    public AsyncExecution(
+            String executionArn,
+            LambdaClient lambdaClient,
+            Class<O> outputType,
+            Duration pollInterval,
+            Duration timeout) {
+        this.executionArn = executionArn;
+        this.lambdaClient = lambdaClient;
+        this.outputType = outputType;
+        this.pollInterval = pollInterval;
+        this.timeout = timeout;
+        this.processor = new HistoryEventProcessor();
+    }
+
+    /**
+     * Poll execution history until the given condition is met.
+     *
+     * @param condition predicate to test on each poll
+     * @return this execution for chaining
+     * @throws TimeoutException if condition not met within timeout
+     */
+    public AsyncExecution<O> pollUntil(Predicate<AsyncExecution<O>> condition) throws TimeoutException {
+        var startTime = Instant.now();
+
+        while (Duration.between(startTime, Instant.now()).compareTo(timeout) < 0) {
+            refreshHistory();
+
+            if (condition.test(this)) {
+                return this;
+            }
+
+            try {
+                Thread.sleep(pollInterval.toMillis());
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                throw new RuntimeException("Polling interrupted", e);
+            }
+        }
+
+        throw new TimeoutException("Condition not met within timeout of " + timeout);
+    }
+
+    /**
+     * Poll until execution completes and return the final result.
+     *
+     * @return test result with execution status and output
+     * @throws TimeoutException if execution doesn't complete within timeout
+     */
+    public TestResult<O> pollUntilComplete() throws TimeoutException {
+        pollUntil(AsyncExecution::isComplete);
+        return currentResult;
+    }
+
+    /**
+     * Check if execution has completed (succeeded or failed).
+     */
+    public boolean isComplete() {
+        if (currentHistory == null) {
+            return false;
+        }
+        return currentHistory.stream()
+                .anyMatch(e -> {
+                    var eventType = e.eventTypeAsString();
+                    return "ExecutionSucceeded".equals(eventType) || "ExecutionFailed".equals(eventType);
+                });
+    }
+
+    /**
+     * Check if an operation with the given name exists.
+     */
+    public boolean hasOperation(String name) {
+        if (currentResult == null) {
+            return false;
+        }
+        return currentResult.getOperations().stream()
+                .anyMatch(op -> name.equals(op.getName()));
+    }
+
+    /**
+     * Check if a callback operation with the given name exists and is started.
+     */
+    public boolean hasCallback(String name) {
+        if (currentHistory == null) {
+            return false;
+        }
+        // Look for CallbackStarted event with this name
+        return currentHistory.stream()
+                .anyMatch(e -> name.equals(e.name()) && "CallbackStarted".equals(e.eventTypeAsString()));
+    }
+
+    /**
+     * Get the callback ID for a callback operation.
+     *
+     * @param operationName name of the callback operation
+     * @return callback ID
+     * @throws IllegalStateException if no callback found for operation
+     */
+    public String getCallbackId(String operationName) {
+        if (currentResult == null) {
+            throw new IllegalStateException("No history available - call pollUntil first");
+        }
+        
+        var operation = currentResult.getOperations().stream()
+                .filter(op -> operationName.equals(op.getName()))
+                .findFirst()
+                .orElseThrow(() -> new IllegalStateException(
+                        "No operation found with name: " + operationName));
+        
+        var callbackDetails = operation.getCallbackDetails();
+        if (callbackDetails == null || callbackDetails.callbackId() == null) {
+            throw new IllegalStateException(
+                    "Operation '" + operationName + "' is not a callback or has no callback ID");
+        }
+        
+        return callbackDetails.callbackId();
+    }
+
+    /**
+     * Get details for a specific operation.
+     */
+    public TestOperation getOperation(String name) {
+        if (currentResult == null) {
+            throw new IllegalStateException("No history available - call pollUntil first");
+        }
+        return currentResult.getOperation(name);
+    }
+
+    /**
+     * Get all operations in the execution.
+     */
+    public List<TestOperation> getOperations() {
+        if (currentResult == null) {
+            throw new IllegalStateException("No history available - call pollUntil first");
+        }
+        return currentResult.getOperations();
+    }
+
+    /**
+     * Get current execution status.
+     */
+    public ExecutionStatus getStatus() {
+        if (currentResult == null) {
+            return ExecutionStatus.PENDING;
+        }
+        return currentResult.getStatus();
+    }
+
+    /**
+     * Get the execution ARN.
+     */
+    public String getExecutionArn() {
+        return executionArn;
+    }
+
+    private void refreshHistory() {
+        try {
+            var request = GetDurableExecutionHistoryRequest.builder()
+                    .durableExecutionArn(executionArn)
+                    .includeExecutionData(true)
+                    .build();
+            var response = lambdaClient.getDurableExecutionHistory(request);
+            this.currentHistory = response.events();
+            this.currentResult = processor.processEvents(currentHistory, outputType);
+        } catch (ResourceNotFoundException e) {
+            // Execution doesn't exist yet - this can happen immediately after async invoke
+            // Leave currentHistory as null, pollUntil will retry
+            this.currentHistory = null;
+            this.currentResult = null;
+        }
+    }
+}

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

@@ -115,6 +115,48 @@ public TestResult<O> run(I input) {
         }
     }
 
+    /**
+     * Start an asynchronous execution and return a handle for incremental polling.
+     * Use this for callback-based tests where you need to interact with the execution
+     * while it's running.
+     *
+     * @param input the input to the function
+     * @return execution handle for polling and inspection
+     */
+    public AsyncExecution<O> startAsync(I input) {
+        try {
+            // Serialize input
+            var serDes = new JacksonSerDes();
+            var inputJson = serDes.serialize(input);
+
+            // Invoke function with EVENT type (async)
+            var invokeRequest = InvokeRequest.builder()
+                    .functionName(functionArn)
+                    .invocationType(InvocationType.EVENT)
+                    .payload(SdkBytes.fromUtf8String(inputJson))
+                    .build();
+
+            var response = lambdaClient.invoke(invokeRequest);
+
+            // Extract execution ARN from response
+            var executionArn = response.durableExecutionArn();
+            if (executionArn == null) {
+                throw new RuntimeException("No durable execution ARN returned - function may not be durable");
+            }
+
+            // Give the execution a moment to initialize before returning
+            // This prevents immediate polling from failing with "execution does not exist"
+            Thread.sleep(100);
+
+            return new AsyncExecution<>(executionArn, lambdaClient, outputType, pollInterval, timeout);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new RuntimeException("Interrupted while starting async execution", e);
+        } catch (Exception e) {
+            throw new RuntimeException("Function invocation failed", e);
+        }
+    }
+
     public TestOperation getOperation(String name) {
         if (lastResult == null) {
             throw new IllegalStateException("No execution has been run yet");