From 2a76fae9ca35b4ad25ec9263a30e52cb9628d1cb Mon Sep 17 00:00:00 2001 From: DeepaliTandale <18361446+DeepaliTandale@users.noreply.github.com> Date: Mon, 15 Jun 2026 14:16:51 +0100 Subject: [PATCH 1/5] first commit --- .../aws-lambda-durable-functions/SKILL.md | 49 ++++ .../references/advanced-error-handling.md | 59 +++- .../references/advanced-patterns.md | 219 ++++++++++++++ .../references/concurrent-operations.md | 241 ++++++++++++++++ .../references/error-handling.md | 220 ++++++++++++++ .../references/getting-started.md | 204 +++++++++++++ .../references/replay-model-rules.md | 144 ++++++++++ .../references/step-operations.md | 165 +++++++++++ .../references/testing-patterns.md | 269 ++++++++++++++++++ .../references/wait-operations.md | 261 +++++++++++++++++ 10 files changed, 1830 insertions(+), 1 deletion(-) diff --git a/plugins/aws-serverless/skills/aws-lambda-durable-functions/SKILL.md b/plugins/aws-serverless/skills/aws-lambda-durable-functions/SKILL.md index 346edc0e..4ba1519a 100644 --- a/plugins/aws-serverless/skills/aws-lambda-durable-functions/SKILL.md +++ b/plugins/aws-serverless/skills/aws-lambda-durable-functions/SKILL.md @@ -24,6 +24,7 @@ Before using AWS Lambda durable functions, verify: 2. **Runtime environment** is ready: - For TypeScript/JavaScript: Node.js 22+ (`node --version`) - For Python: Python 3.11+ (`python --version`. Note that currently only Lambda runtime environments 3.13+ come with the Durable Execution SDK pre-installed. 3.11 is the min supported Python version by the Durable SDK itself, however, you could use OCI to bring your own container image with your own Python runtime + Durable SDK.) + - For Java: Java 17+ (`java --version`) 3. **Deployment capability** exists (one of): - AWS SAM CLI (`sam --version`) 1.153.1 or higher @@ -40,6 +41,7 @@ Override syntax: - "use Python" → Generate Python code - "use JavaScript" → Generate JavaScript code +- "use Java" → Generate Java code When not specified, ALWAYS use TypeScript @@ -90,6 +92,24 @@ pip install aws-durable-execution-sdk-python pip install aws-durable-execution-sdk-python-testing ``` +**For Java (Maven):** + +Add the dependencies to your `pom.xml`, pinning to a specific released `VERSION` (see [Maven Central](https://mvnrepository.com/artifact/software.amazon.lambda.durable/aws-durable-execution-sdk-java)): + +```xml + + software.amazon.lambda.durable + aws-durable-execution-sdk-java + VERSION + + + software.amazon.lambda.durable + aws-durable-execution-sdk-java-testing + VERSION + test + +``` + ## When to Load Reference Files Load the appropriate reference file based on what the user is working on: @@ -132,6 +152,21 @@ def handler(event: dict, context: DurableContext) -> dict: return result ``` +**Java:** + +```java +import software.amazon.lambda.durable.DurableHandler; +import software.amazon.lambda.durable.DurableContext; + +public class MyHandler extends DurableHandler { + @Override + public MyOutput handleRequest(MyInput event, DurableContext ctx) { + var result = ctx.step("process", ProcessResult.class, stepCtx -> processData(event)); + return result; + } +} +``` + ### Critical Rules 1. **All non-deterministic code MUST be in steps** (Date.now, Math.random, API calls) @@ -148,6 +183,19 @@ The Python SDK differs from TypeScript in several key areas: - **Exceptions**: `ExecutionError` (permanent), `InvocationError` (transient), `CallbackError` (callback failures) - **Testing**: Use `DurableFunctionTestRunner` class directly - instantiate with handler, use context manager, call `run(input=...)` +### Java API Differences + +The Java SDK uses a class-based, type-safe approach: + +- **Handler**: Extend `DurableHandler` and implement `handleRequest(TInput, DurableContext)` +- **Steps**: `ctx.step("name", ResultType.class, stepCtx -> operation())` — a name and result type are always required +- **Generic types**: Use `TypeToken` for generics like `List`: `ctx.step("name", new TypeToken>() {}, stepCtx -> ...)` +- **Wait**: `ctx.wait("name", Duration.ofSeconds(n))` +- **Async operations**: `stepAsync()`, `invokeAsync()`, `waitAsync()`, `mapAsync()` return `DurableFuture`; combine with `DurableFuture.allOf(...)` / `DurableFuture.anyOf(...)` +- **Parallel**: `ctx.parallel("name", config)` returns a `ParallelDurableFuture` (AutoCloseable); register branches with `parallel.branch(...)` +- **Exceptions**: `StepFailedException`, `StepInterruptedException`, `CallbackTimeoutException`, `CallbackFailedException`, `WaitForConditionFailedException`, `InvokeFailedException`, `InvokeTimedOutException`, `UnrecoverableDurableExecutionException`, with `DurableExecutionException` as the base class +- **Testing**: Use `LocalDurableTestRunner` (local) or `CloudDurableTestRunner` (cloud) from the testing SDK + ### Invocation Requirements Durable functions **require qualified ARNs** (version, alias, or `$LATEST`): @@ -201,4 +249,5 @@ Access to sensitive data (like Lambda and API Gateway logs) is **not** enabled b - [AWS Lambda durable functions Documentation](https://docs.aws.amazon.com/lambda/latest/dg/durable-functions.html) - [JavaScript SDK Repository](https://github.com/aws/aws-durable-execution-sdk-js) - [Python SDK Repository](https://github.com/aws/aws-durable-execution-sdk-python) +- [Java SDK Repository](https://github.com/aws/aws-durable-execution-sdk-java) - [IAM Policy Reference](https://docs.aws.amazon.com/aws-managed-policy/latest/reference/AWSLambdaBasicDurableExecutionRolePolicy.html) diff --git a/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/advanced-error-handling.md b/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/advanced-error-handling.md index b9c9b26e..b6ebc138 100644 --- a/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/advanced-error-handling.md +++ b/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/advanced-error-handling.md @@ -8,7 +8,7 @@ Advanced error handling patterns for durable functions, including timeout handli **Implementation approach:** -1. Use `waitForCallback` (TypeScript) or `wait_for_callback` (Python) with a timeout configuration set in the config argument +1. Use `waitForCallback` (TypeScript), `wait_for_callback` (Python), or `ctx.waitForCallback` (Java) with a timeout configuration set in the config argument 2. Wrap in try-catch to handle timeout errors 3. Check if the error is a timeout 4. Implement fallback logic in a step (e.g., escalate to manager, use default value, retry with different parameters) @@ -34,6 +34,30 @@ Advanced error handling patterns for durable functions, including timeout handli **Important limitation:** In TypeScript, native setTimeout (and patterns like Promise.race using it) will fail during execution replays. To create a reliable timeout that persists across execution (expands over multi invocations), always use the timeout parameter provided by waitForCallback or waitForCondition +**Java equivalent — `DurableFuture.anyOf`:** +Java has no `Promise.race`, but `DurableFuture.anyOf(...)` waits for the first of several async durable operations to complete, returning that operation's result. For reliable cross-invocation timeouts that persist across replays, always use the timeout configuration in `CallbackConfig` (via `WaitForCallbackConfig`) or `WaitForConditionConfig` rather than any in-process timer. + +```java +import software.amazon.lambda.durable.DurableFuture; + +// Race multiple async operations - first to complete wins +var primary = ctx.stepAsync("primary-api", Result.class, s -> callPrimaryAPI()); +var backup = ctx.stepAsync("backup-api", Result.class, s -> callBackupAPI()); + +// Block until the first one completes +DurableFuture.anyOf(primary, backup); + +// Read whichever finished first +Result result; +try { + result = primary.get(); + ctx.getLogger().info("Primary API completed first"); +} catch (Exception e) { + result = backup.get(); + ctx.getLogger().info("Backup API completed first"); +} +``` + ## Conditional Retry Based on Error Type **Pattern:** Retry operations selectively based on the type of error encountered. @@ -113,3 +137,36 @@ In TypeScript, native setTimeout (and patterns like Promise.race using it) will - Callback timeouts - external system didn't respond in time - External system delays - service is slow or unresponsive - Long-running operations - operation exceeded expected duration + +## Java Exception Types + +The Java SDK surfaces typed exceptions you can catch to drive fallback logic: + +| Exception | Category | Use case | +| --------------------------------- | ----------------- | ------------------------------------------------------- | +| `CallbackTimeoutException` | Timeout | Callback didn't complete within its timeout | +| `CallbackFailedException` | Callback failure | Callback failed or was explicitly rejected | +| `WaitForConditionFailedException` | Condition failure | Condition check failed or max polling attempts exceeded | +| `StepFailedException` | Permanent | Step failed after exhausting retries (business error) | +| `StepInterruptedException` | Transient | Step interrupted before completion (can retry) | +| `DurableExecutionException` | Base | Base class for all SDK exceptions | + +```java +import software.amazon.lambda.durable.config.WaitForCallbackConfig; +import software.amazon.lambda.durable.config.CallbackConfig; +import software.amazon.lambda.durable.exception.CallbackTimeoutException; +import software.amazon.lambda.durable.exception.CallbackFailedException; + +try { + var decision = ctx.waitForCallback("wait-approval", Decision.class, + (callbackId, s) -> sendApproval(callbackId), + WaitForCallbackConfig.builder() + .callbackConfig(CallbackConfig.builder().timeout(Duration.ofHours(24)).build()) + .build()); +} catch (CallbackTimeoutException e) { + // Fallback in a step so it is checkpointed + ctx.step("escalate", Void.class, s -> { escalateToManager(); return null; }); +} catch (CallbackFailedException e) { + ctx.getLogger().error("Callback failed: {}", e.getMessage()); +} +``` diff --git a/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/advanced-patterns.md b/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/advanced-patterns.md index c0983eee..3ae66998 100644 --- a/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/advanced-patterns.md +++ b/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/advanced-patterns.md @@ -96,6 +96,48 @@ def handler(event: dict, context: DurableContext) -> str: context.logger.debug('Tool result added', extra={'tool': tool['name']}) ``` +**Java:** + +```java +public class AgentHandler extends DurableHandler { + @Override + public String handleRequest(AgentRequest event, DurableContext ctx) { + ctx.getLogger().info("Starting AI agent with prompt: {}", event.getPrompt()); + var messages = new ArrayList<>(List.of( + Map.of("role", "user", "content", event.getPrompt()))); + + while (true) { + // Invoke AI model with reasoning + var result = ctx.step("invoke-model", ModelResult.class, stepCtx -> { + stepCtx.getLogger().info("Invoking AI model. Message count: {}", messages.size()); + return invokeAIModel(messages); + }); + + // Log AI's reasoning + if (result.getReasoning() != null) { + ctx.getLogger().debug("AI reasoning: {}", result.getReasoning()); + } + + // If no tool needed, return response + if (result.getTool() == null) { + ctx.getLogger().info("AI agent completed - no tool needed"); + return result.getResponse(); + } + + // Execute tool with dynamic step naming + var toolResult = ctx.step("execute-tool-" + result.getTool().getName(), String.class, + stepCtx -> { + stepCtx.getLogger().info("Executing tool: {}", result.getTool().getName()); + return executeTool(result.getTool(), result.getResponse()); + }); + + messages.add(Map.of("role", "assistant", "content", toolResult)); + ctx.getLogger().debug("Tool result added. Tool: {}", result.getTool().getName()); + } + } +} +``` + ## Step Semantics Deep Dive ### AtMostOncePerRetry vs AtLeastOncePerRetry @@ -130,6 +172,30 @@ await context.step( ); ``` +**Java:** + +```java +import software.amazon.lambda.durable.config.StepConfig; +import software.amazon.lambda.durable.config.StepSemantics; + +// AT_MOST_ONCE_PER_RETRY - the START checkpoint is awaited before user code runs. +// If interrupted, the step is not silently re-run within the same attempt. +// Use semanticsPerRetry(...) so interrupted steps still follow the retry strategy. +ctx.step("update-database", UpdateResult.class, + stepCtx -> updateUserRecord(userId, data), + StepConfig.builder() + .semanticsPerRetry(StepSemantics.AT_MOST_ONCE_PER_RETRY) + .build()); + +// AT_LEAST_ONCE_PER_RETRY (default) - may execute multiple times per retry attempt. +// Use when idempotency is handled externally. +ctx.step("send-notification", Void.class, + stepCtx -> { sendEmail(email, message); return null; }, + StepConfig.builder() + .semanticsPerRetry(StepSemantics.AT_LEAST_ONCE_PER_RETRY) + .build()); +``` + **When to use each:** | Semantic | Use When | Example Operations | @@ -165,6 +231,23 @@ const results = await context.map( // 3. 20% of items fail (toleratedFailurePercentage reached) ``` +**Java:** + +```java +// CompletionConfig is a record; use the canonical constructor to combine constraints. +// Percentage is a 0.0-1.0 fraction. +var results = ctx.map("process-items", items, Result.class, + (item, index, childCtx) -> childCtx.step("p-" + index, Result.class, s -> process(item)), + MapConfig.builder() + .completionConfig(new CompletionConfig(8, 2, 0.20)) + .build()); + +// Execution stops when ANY of these conditions is met: +// 1. 8 successful items (minSuccessful reached) +// 2. 2 failures occur (toleratedFailureCount reached) +// 3. 20% of items fail (toleratedFailurePercentage reached) +``` + ### Understanding Stop Conditions **Example with 10 items:** @@ -199,6 +282,31 @@ const results = await context.map( // 1 item not processed, but completion policy satisfied ``` +**Java:** + +```java +var items = IntStream.range(0, 10).boxed().toList(); + +var results = ctx.map("process", items, Result.class, + (item, index, childCtx) -> childCtx.step("p-" + index, Result.class, s -> process(item)), + MapConfig.builder() + .maxConcurrency(3) + .completionConfig(new CompletionConfig(7, 3, null)) + .build()); + +// Scenario 1: 7 successes, 0 failures +// ✅ Stops after 7th success (minSuccessful reached) +// Remaining 3 items are not processed + +// Scenario 2: 5 successes, 3 failures +// ❌ Stops after 3rd failure (toleratedFailureCount reached) +// Remaining 2 items are not processed; results.allSucceeded() is false + +// Scenario 3: 7 successes, 2 failures +// ✅ Stops after 7th success (minSuccessful reached) +// 1 item not processed, but completion policy satisfied +``` + ### Early Termination Pattern Use completion policies for early termination when searching: @@ -227,6 +335,25 @@ if (results.successCount > 0) { } ``` +**Java:** + +```java +// Stop after finding the first match +var results = ctx.map("find-match", candidates, Match.class, + (candidate, index, childCtx) -> childCtx.step("check-" + index, Match.class, + s -> checkMatch(candidate)), + MapConfig.builder() + .completionConfig(CompletionConfig.minSuccessful(1)) // Stop after first success + .build()); + +// Only one item processed (assuming the first succeeds) +var matches = results.succeeded(); +if (!matches.isEmpty()) { + var match = matches.get(0); + ctx.getLogger().info("Found match: {}", match); +} +``` + ## Advanced Error Handling For timeout handling (waitForCallback, Promise.race), conditional retries, and circuit breaker patterns, see [advanced-error-handling.md](advanced-error-handling.md). @@ -267,6 +394,44 @@ const result = await context.step( console.log(result.createdAt instanceof Date); // true ``` +**Java:** + +```java +import software.amazon.lambda.durable.serde.SerDes; +import software.amazon.lambda.durable.TypeToken; +import software.amazon.lambda.durable.config.StepConfig; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule; + +// Jackson handles java.time types (Instant, etc.) via JavaTimeModule. +// SerDes is non-generic: serialize(Object) and deserialize(String, TypeToken). +public class DateAwareSerDes implements SerDes { + private final ObjectMapper mapper = new ObjectMapper().registerModule(new JavaTimeModule()); + + @Override + public String serialize(Object value) { + try { + return mapper.writeValueAsString(value); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + + @Override + public T deserialize(String data, TypeToken typeToken) { + try { + return mapper.readValue(data, mapper.constructType(typeToken.getType())); + } catch (Exception e) { + throw new RuntimeException(e); + } + } +} + +var result = ctx.step("create-user", User.class, + stepCtx -> new User("Alice", "alice@example.com", Instant.now(), Instant.now()), + StepConfig.builder().serDes(new DateAwareSerDes()).build()); +``` + ### Complex Object Graphs **TypeScript:** @@ -309,6 +474,22 @@ const result = await context.step( ); ``` +**Java:** + +```java +// Nested object graphs serialize automatically with the default Jackson-based +// SerDes - records (or POJOs) compose without per-class registration. +public record Customer(String id, String name) {} +public record OrderItem(String sku, int quantity) {} +public record Order(String id, List items, Customer customer) {} + +var result = ctx.step("process-order", Order.class, stepCtx -> { + var customer = new Customer("CUST-123", "Alice"); + var items = List.of(new OrderItem("SKU-001", 2), new OrderItem("SKU-002", 1)); + return new Order("ORD-456", items, customer); +}); +``` + ## Nested Workflows ### Parent-Child Workflow Pattern @@ -366,6 +547,44 @@ export const worker = withDurableExecution( ); ``` +**Java:** + +```java +import software.amazon.lambda.durable.DurableFuture; + +// Parent orchestrator +public class OrchestratorHandler extends DurableHandler> { + @Override + public List handleRequest(BatchEvent event, DurableContext ctx) { + var childArn = System.getenv("CHILD_FUNCTION_ARN"); + + // Invoke child workflows concurrently with invokeAsync (durable operations + // cannot be nested inside a step). + var f1 = ctx.invokeAsync("batch-1", childArn, event.getBatches().get(0), BatchResult.class); + var f2 = ctx.invokeAsync("batch-2", childArn, event.getBatches().get(1), BatchResult.class); + + // Wait for all child invocations to complete + DurableFuture.allOf(f1, f2); + + return List.of(f1.get(), f2.get()); + } +} + +// Child worker +public class WorkerHandler extends DurableHandler> { + @Override + public List handleRequest(BatchInput event, DurableContext ctx) { + var items = event.getBatch().getItems(); + + var results = ctx.map("process-items", items, Result.class, + (item, index, childCtx) -> childCtx.step("process-" + index, Result.class, + stepCtx -> processItem(item))); + + return results.results(); + } +} +``` + ## Best Practices Summary 1. **Dynamic Step Naming**: Use template literals for dynamic operation names diff --git a/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/concurrent-operations.md b/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/concurrent-operations.md index 4df17b9d..e40021a4 100644 --- a/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/concurrent-operations.md +++ b/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/concurrent-operations.md @@ -60,6 +60,27 @@ results.throw_if_error() all_results = results.get_results() ``` +**Java:** + +```java +import software.amazon.lambda.durable.config.MapConfig; +import software.amazon.lambda.durable.config.CompletionConfig; + +List items = List.of(1, 2, 3, 4, 5); + +// MapFunction signature is (item, index, childContext) +var results = ctx.map("process-items", items, Result.class, + (item, index, childCtx) -> childCtx.step("process-" + index, Result.class, + stepCtx -> processItem(item)), + MapConfig.builder() + .maxConcurrency(3) + .completionConfig(CompletionConfig.minSuccessful(4)) + .build()); + +// results() returns the ordered List (nulls for failed/not-started items) +var allResults = results.results(); +``` + ## Parallel Operations Run heterogeneous operations concurrently: @@ -113,6 +134,36 @@ results = context.parallel( user, orders, preferences = results.get_results() ``` +**Java:** + +```java +import software.amazon.lambda.durable.DurableFuture; +import software.amazon.lambda.durable.ParallelDurableFuture; +import software.amazon.lambda.durable.config.ParallelConfig; + +// ParallelDurableFuture is AutoCloseable; try-with-resources guarantees +// join() runs even if an exception occurs. +var config = ParallelConfig.builder().maxConcurrency(3).build(); +var parallel = ctx.parallel("parallel-ops", config); + +DurableFuture userF; +DurableFuture> ordersF; +DurableFuture prefsF; + +try (parallel) { + userF = parallel.branch("fetch-user", User.class, + branchCtx -> branchCtx.step("fetch-user-step", User.class, s -> fetchUser(userId))); + ordersF = parallel.branch("fetch-orders", new TypeToken>() {}, + branchCtx -> branchCtx.step("fetch-orders-step", new TypeToken>() {}, s -> fetchOrders(userId))); + prefsF = parallel.branch("fetch-preferences", Preferences.class, + branchCtx -> branchCtx.step("fetch-prefs-step", Preferences.class, s -> fetchPreferences(userId))); +} // join() called here via close() + +var user = userF.get(); +var orders = ordersF.get(); +var preferences = prefsF.get(); +``` + ## Completion Policies ### Minimum Successful @@ -134,6 +185,16 @@ const results = await context.map( ); ``` +**Java:** + +```java +var results = ctx.map("process-batch", items, Result.class, + (item, index, childCtx) -> childCtx.step("process-" + index, Result.class, s -> process(item)), + MapConfig.builder() + .completionConfig(CompletionConfig.minSuccessful(8)) // Need at least 8 successes + .build()); +``` + ### Tolerated Failures Allow a specific number of failures: @@ -153,6 +214,16 @@ const results = await context.map( ); ``` +**Java:** + +```java +var results = ctx.map("process-batch", items, Result.class, + (item, index, childCtx) -> childCtx.step("process-" + index, Result.class, s -> process(item)), + MapConfig.builder() + .completionConfig(CompletionConfig.toleratedFailureCount(2)) // Allow up to 2 failures + .build()); +``` + ### Tolerated Failure Percentage Allow a percentage of failures: @@ -187,6 +258,17 @@ results = context.map( ) ``` +**Java:** + +```java +// Java expresses the percentage as a 0.0-1.0 fraction (0.1 = 10%) +var results = ctx.map("process-batch", items, Result.class, + (item, index, childCtx) -> childCtx.step("process-" + index, Result.class, s -> process(item)), + MapConfig.builder() + .completionConfig(CompletionConfig.toleratedFailurePercentage(0.1)) // Allow up to 10% failures + .build()); +``` + ## Batch Result Handling ### Check Status @@ -204,6 +286,19 @@ console.log(results.failureCount); // Failed items console.log(results.hasFailure()); // Boolean ``` +**Java:** + +```java +var results = ctx.map("process", items, Result.class, + (item, index, childCtx) -> childCtx.step("p-" + index, Result.class, s -> process(item))); + +System.out.println(results.size()); // Total items +System.out.println(results.allSucceeded()); // true if no failures/not-started +System.out.println(results.succeeded().size()); // Successful items +System.out.println(results.failed().size()); // Failed items (List) +System.out.println(results.completionReason()); // ConcurrencyCompletionStatus +``` + ### Get Results **TypeScript:** @@ -230,6 +325,32 @@ const all = results.all.map(item => ({ })); ``` +**Java:** + +```java +import software.amazon.lambda.durable.model.MapResult.MapResultItem; +import software.amazon.lambda.durable.model.MapResult.MapError; + +// All results in input order (null for failed/not-started items) +List allResults = results.results(); + +// Successful results only +List successful = results.succeeded(); + +// Errors from failed items +List failures = results.failed(); + +// Inspect each item by index with its status +for (int i = 0; i < results.size(); i++) { + MapResultItem item = results.getItem(i); + switch (item.status()) { + case SUCCEEDED -> { var value = item.result(); /* use value */ } + case FAILED -> ctx.getLogger().error("Item {} failed: {}", i, item.error().errorMessage()); + case SKIPPED -> ctx.getLogger().info("Item {} was skipped", i); + } +} +``` + ### Error Handling **TypeScript:** @@ -249,6 +370,32 @@ if (results.hasFailure()) { } ``` +**Java:** + +```java +import software.amazon.lambda.durable.model.MapResult.MapResultItem; + +var results = ctx.map("process", items, Result.class, + (item, index, childCtx) -> childCtx.step("process-" + index, Result.class, s -> process(item)), + MapConfig.builder().completionConfig(CompletionConfig.minSuccessful(3)).build()); + +// Individual item failures don't throw - inspect the result explicitly +if (!results.allSucceeded()) { + ctx.getLogger().error("Some items failed. Error count: {}", results.failed().size()); + + // Collect failed indices to rebuild the retry input + var failedItems = new ArrayList(); + for (int i = 0; i < results.size(); i++) { + if (results.getItem(i).status() == MapResultItem.Status.FAILED) { + failedItems.add(items.get(i)); + } + } + + ctx.map("retry-failed", failedItems, Result.class, + (item, index, childCtx) -> childCtx.step("retry-" + index, Result.class, s -> process(item))); +} +``` + ## Concurrency Control ### Fixed Concurrency @@ -264,6 +411,14 @@ const results = await context.map( ); ``` +**Java:** + +```java +var results = ctx.map("process", items, Result.class, + (item, index, childCtx) -> childCtx.step("process-" + index, Result.class, s -> process(item)), + MapConfig.builder().maxConcurrency(5).build()); // Process 5 items at a time +``` + ### Dynamic Concurrency Adjust based on item characteristics: @@ -291,6 +446,21 @@ const results = await context.map( ); ``` +**Java:** + +```java +var results = ctx.map("process", items, Result.class, + (item, index, childCtx) -> { + // Heavy items get their own processing + if (item.getSize() > 1000) { + return childCtx.step("heavy-" + index, Result.class, s -> processHeavy(item)); + } + // Light items can be batched + return childCtx.step("light-" + index, Result.class, s -> processLight(item)); + }, + MapConfig.builder().maxConcurrency(10).build()); +``` + ## Advanced Patterns ### Map with Callbacks @@ -318,6 +488,28 @@ const results = await context.map( ); ``` +**Java:** + +```java +import software.amazon.lambda.durable.config.WaitForCallbackConfig; +import software.amazon.lambda.durable.config.CallbackConfig; + +var results = ctx.map("process-with-approval", items, ApprovalResult.class, + (item, index, childCtx) -> { + var processed = childCtx.step("process", ProcessResult.class, + s -> process(item)); + + var approved = childCtx.waitForCallback("approval", ApprovalData.class, + (callbackId, s) -> sendApproval(item, callbackId), + WaitForCallbackConfig.builder() + .callbackConfig(CallbackConfig.builder().timeout(Duration.ofHours(24)).build()) + .build()); + + return new ApprovalResult(processed, approved); + }, + MapConfig.builder().maxConcurrency(3).build()); +``` + ### Nested Map Operations **TypeScript:** @@ -338,6 +530,18 @@ const results = await context.map( ); ``` +**Java:** + +```java +var results = ctx.map("process-batches", batches, BatchResult.class, + (batch, batchIndex, batchCtx) -> { + var itemResults = batchCtx.map("batch-" + batchIndex, batch.getItems(), ItemResult.class, + (item, itemIndex, itemCtx) -> itemCtx.step("item-" + itemIndex, ItemResult.class, + s -> process(item))); + return new BatchResult(itemResults.results()); + }); +``` + ### Map with Child Contexts **TypeScript:** @@ -365,6 +569,18 @@ const results = await context.map( ); ``` +**Java:** + +```java +var results = ctx.map("complex-process", items, ProcessResult.class, + (item, index, childCtx) -> childCtx.runInChildContext("item-" + index, ProcessResult.class, inner -> { + var validated = inner.step("validate", Validated.class, s -> validate(item)); + inner.wait("item-delay", Duration.ofSeconds(1)); + return inner.step("process", ProcessResult.class, s -> process(validated)); + }), + MapConfig.builder().maxConcurrency(5).build()); +``` + ## Performance Optimization ### Batch Size Selection @@ -387,6 +603,20 @@ const results = await context.map( ); ``` +**Java:** + +```java +// Small items: Higher concurrency +var smallResults = ctx.map("small-items", smallItems, Result.class, + (item, index, childCtx) -> childCtx.step("p-" + index, Result.class, s -> process(item)), + MapConfig.builder().maxConcurrency(20).build()); + +// Large items: Lower concurrency +var largeResults = ctx.map("large-items", largeItems, Result.class, + (item, index, childCtx) -> childCtx.step("p-" + index, Result.class, s -> process(item)), + MapConfig.builder().maxConcurrency(3).build()); +``` + ### Early Termination Use completion policies to stop early: @@ -406,6 +636,17 @@ const results = await context.map( ); ``` +**Java:** + +```java +var results = ctx.map("find-match", candidates, Match.class, + (candidate, index, childCtx) -> childCtx.step("check-" + index, Match.class, + s -> checkMatch(candidate)), + MapConfig.builder() + .completionConfig(CompletionConfig.minSuccessful(1)) // Stop after first success + .build()); +``` + ## Best Practices 1. **Set appropriate maxConcurrency** based on downstream system capacity diff --git a/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/error-handling.md b/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/error-handling.md index 713578f3..86afd149 100644 --- a/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/error-handling.md +++ b/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/error-handling.md @@ -55,6 +55,32 @@ result = context.step( ) ``` +**Java:** + +```java +import java.time.Duration; +import software.amazon.lambda.durable.config.StepConfig; +import software.amazon.lambda.durable.retry.RetryStrategies; +import software.amazon.lambda.durable.retry.JitterStrategy; + +// Exponential backoff with jitter +var result = ctx.step("api-call", ApiResponse.class, stepCtx -> callAPI(), + StepConfig.builder() + .retryStrategy(RetryStrategies.exponentialBackoff( + 5, // maxAttempts (including initial attempt) + Duration.ofSeconds(1), // initialDelay + Duration.ofSeconds(60), // maxDelay + 2.0, // backoffRate + JitterStrategy.FULL)) + .build()); + +// Fixed delay +var simpleResult = ctx.step("simple-retry", Result.class, stepCtx -> operation(), + StepConfig.builder() + .retryStrategy(RetryStrategies.fixedDelay(3, Duration.ofSeconds(5))) + .build()); +``` + ## Custom Retry Logic **TypeScript:** @@ -100,6 +126,28 @@ def custom_retry(error: Exception, attempt: int) -> RetryDecision: return RetryDecision(should_retry=False) ``` +**Java:** + +```java +import software.amazon.lambda.durable.retry.RetryDecision; + +var result = ctx.step("custom-retry", Result.class, stepCtx -> riskyOperation(), + StepConfig.builder() + .retryStrategy((error, attemptCount) -> { + // Don't retry client errors (4xx) + if (error instanceof HttpException he + && he.getStatusCode() >= 400 && he.getStatusCode() < 500) { + return RetryDecision.fail(); + } + // Retry server errors with exponential backoff + if (attemptCount < 5) { + return RetryDecision.retry(Duration.ofSeconds((long) Math.pow(2, attemptCount))); + } + return RetryDecision.fail(); + }) + .build()); +``` + ## Error Classification ### Retryable vs Non-Retryable @@ -137,6 +185,27 @@ retry_config = RetryStrategyConfig( ) ``` +**Java:** + +```java +import software.amazon.lambda.durable.retry.RetryDecision; + +// The Java SDK's built-in strategies have no "retryable types" option; +// inspect the error type inside a custom strategy instead. +var result = ctx.step("selective-retry", Result.class, stepCtx -> operation(), + StepConfig.builder() + .retryStrategy((error, attemptCount) -> { + boolean retryable = error instanceof NetworkException + || error instanceof TimeoutException; + if (retryable && attemptCount < 3) { + return RetryDecision.retry(Duration.ofSeconds((long) Math.pow(2, attemptCount))); + } + // ValidationException (and any non-listed type) won't be retried + return RetryDecision.fail(); + }) + .build()); +``` + ## Saga Pattern Implement compensating transactions for distributed workflows: @@ -232,6 +301,51 @@ def handler(event: dict, context: DurableContext) -> dict: raise error ``` +**Java:** + +```java +import software.amazon.lambda.durable.exception.DurableExecutionException; + +public class OrderHandler extends DurableHandler { + @Override + public OrderResult handleRequest(OrderRequest event, DurableContext ctx) { + var compensations = new ArrayList(); + try { + // Step 1: Reserve inventory + var reservation = ctx.step("reserve-inventory", Reservation.class, + s -> inventoryService.reserve(event.getItems())); + compensations.add(() -> ctx.step("cancel-reservation", Void.class, + s -> { inventoryService.cancelReservation(reservation.getId()); return null; })); + + // Step 2: Charge payment + var payment = ctx.step("charge-payment", Payment.class, + s -> paymentService.charge(event.getPaymentMethod(), event.getAmount())); + compensations.add(() -> ctx.step("refund-payment", Void.class, + s -> { paymentService.refund(payment.getId()); return null; })); + + // Step 3: Create shipment + var shipment = ctx.step("create-shipment", Shipment.class, + s -> shippingService.createShipment(event.getAddress(), event.getItems())); + + return new OrderResult(true, shipment.getOrderId()); + } catch (DurableExecutionException e) { + ctx.getLogger().error("Order failed, executing compensations: {}", e.getMessage()); + // Execute compensations in reverse order + Collections.reverse(compensations); + for (var comp : compensations) { + try { + comp.run(); + } catch (DurableExecutionException ce) { + ctx.getLogger().error("Compensation failed: {}", ce.getMessage()); + // Continue with other compensations + } + } + throw e; + } + } +} +``` + ## Unrecoverable Errors Mark errors as unrecoverable to stop execution immediately: @@ -276,6 +390,34 @@ def handler(event: dict, context: DurableContext) -> dict: # Continue processing... ``` +**Java:** + +```java +import software.amazon.lambda.durable.exception.UnrecoverableDurableExecutionException; +import software.amazon.awssdk.services.lambda.model.ErrorObject; + +public class MyHandler extends DurableHandler { + @Override + public MyOutput handleRequest(MyInput event, DurableContext ctx) { + var user = ctx.step("fetch-user", User.class, stepCtx -> { + var u = fetchUser(event.getUserId()); + if (u == null) { + // Terminate execution immediately - no retry. + // A plain RuntimeException would be retried by the step's retry + // strategy; UnrecoverableDurableExecutionException is not. + throw new UnrecoverableDurableExecutionException( + ErrorObject.builder() + .errorType("UserNotFound") + .errorMessage("User not found") + .build()); + } + return u; + }); + // Continue processing... + } +} +``` + The SDK provides these exception types for different failure scenarios: | Exception | Retryable | Use case | @@ -285,6 +427,20 @@ The SDK provides these exception types for different failure scenarios: | `CallbackError` | No | Callback handling failures | | `DurableExecutionsError` | — | Base class for all SDK exceptions | +**Java** SDK exception types: + +| Exception | Retryable | Use case | +| -------------------------------------- | --------- | ------------------------------------------------------- | +| `StepFailedException` | No | Step execution failed (business logic error) | +| `StepInterruptedException` | Yes | Step with AT_MOST_ONCE_PER_RETRY interrupted before completion | +| `CallbackTimeoutException` | No | Callback didn't complete within timeout | +| `CallbackFailedException` | No | Callback failed or was explicitly rejected | +| `WaitForConditionFailedException` | No | Condition check failed or max polling attempts exceeded | +| `InvokeFailedException` | No | Lambda invocation failed | +| `InvokeTimedOutException` | No | Lambda invocation timed out | +| `UnrecoverableDurableExecutionException` | No | Terminate execution immediately, no retry | +| `DurableExecutionException` | — | Base class for all SDK exceptions | + ## Error Determinism Ensure errors are deterministic across replays: @@ -317,6 +473,33 @@ const result = await context.step('validate', async () => { }); ``` +**Java:** + +```java +// Define a deterministic, serializable business exception +public class CustomBusinessException extends RuntimeException { + private final String code; + private final String field; + + public CustomBusinessException(String message, String code, String field) { + super(message); + this.code = code; + this.field = field; + } + + public String getCode() { return code; } + public String getField() { return field; } +} + +var result = ctx.step("validate", ProcessResult.class, stepCtx -> { + if (!isValid(data)) { + // ✅ Deterministic error - same input produces the same failure + throw new CustomBusinessException("Validation failed", "INVALID_DATA", "email"); + } + return processData(data); +}); +``` + ## Circuit Breaker Pattern **TypeScript:** @@ -416,6 +599,43 @@ export const handler = withDurableExecution(async (event, context: DurableContex }); ``` +**Java:** + +```java +import software.amazon.lambda.durable.config.MapConfig; +import software.amazon.lambda.durable.config.CompletionConfig; +import software.amazon.lambda.durable.model.MapResult.MapResultItem; + +public class BatchHandler extends DurableHandler { + @Override + public BatchOutput handleRequest(BatchInput event, DurableContext ctx) { + var results = ctx.map("process-items", event.getItems(), Result.class, + (item, index, childCtx) -> childCtx.step("process-" + index, Result.class, + s -> processItem(item)), + MapConfig.builder() + .completionConfig(CompletionConfig.toleratedFailurePercentage(0.1)) // Allow 10% failures + .build()); + + if (!results.allSucceeded()) { + // Log failures but continue + ctx.getLogger().warn("Some items failed. Failure count: {}", results.failed().size()); + + // Collect failed items for later retry + var failedItems = new ArrayList(); + for (int i = 0; i < results.size(); i++) { + if (results.getItem(i).status() == MapResultItem.Status.FAILED) { + failedItems.add(event.getItems().get(i)); + } + } + + ctx.step("store-failures", Void.class, s -> { storeFailedItems(failedItems); return null; }); + } + + return new BatchOutput(results.succeeded().size(), results.failed().size()); + } +} +``` + ## Best Practices 1. **Use appropriate retry strategies** - exponential backoff for most cases diff --git a/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/getting-started.md b/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/getting-started.md index 3c0eee05..48d9dc6f 100644 --- a/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/getting-started.md +++ b/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/getting-started.md @@ -10,6 +10,7 @@ Override syntax: - "use Python" → Generate Python code - "use JavaScript" → Generate JavaScript code +- "use Java" → Generate Java code When not specified, ALWAYS use TypeScript @@ -66,6 +67,32 @@ def handler(event: dict, context: DurableContext) -> dict: return {'success': True, 'data': result} ``` +**Java:** + +```java +import java.time.Duration; +import software.amazon.lambda.durable.DurableContext; +import software.amazon.lambda.durable.DurableHandler; + +public class MyHandler extends DurableHandler { + @Override + public MyOutput handleRequest(MyInput event, DurableContext ctx) { + // Execute a step with automatic retry + var userData = ctx.step("fetch-user", UserData.class, + stepCtx -> fetchUserFromDB(event.getUserId())); + + // Wait without compute charges + ctx.wait("initial-delay", Duration.ofSeconds(5)); + + // Process in another step + var result = ctx.step("process", ProcessResult.class, + stepCtx -> processUser(userData)); + + return new MyOutput(true, result); + } +} +``` + ## Common Patterns ### Multi-Step Workflow @@ -92,6 +119,28 @@ export const handler = withDurableExecution(async (event, context: DurableContex }); ``` +**Java:** + +```java +public class WorkflowHandler extends DurableHandler { + @Override + public WorkflowOutput handleRequest(WorkflowInput event, DurableContext ctx) { + var validated = ctx.step("validate", ValidatedData.class, + stepCtx -> validateInput(event)); + + var processed = ctx.step("process", ProcessedData.class, + stepCtx -> processData(validated)); + + ctx.wait("cooldown", Duration.ofSeconds(30)); + + ctx.step("notify", Void.class, + stepCtx -> { sendNotification(processed); return null; }); + + return new WorkflowOutput(true); + } +} +``` + ### GenAI Agent (Agentic Loop) **TypeScript:** @@ -135,6 +184,33 @@ def handler(event: dict, context: DurableContext) -> str: messages.append({"role": "assistant", "content": tool_result}) ``` +**Java:** + +```java +public class AgentHandler extends DurableHandler { + @Override + public String handleRequest(AgentInput event, DurableContext ctx) { + var messages = new ArrayList<>(List.of( + Map.of("role", "user", "content", event.getPrompt()))); + + while (true) { + var result = ctx.step("invoke-model", ModelResult.class, + stepCtx -> invokeAIModel(messages)); + + if (result.getTool() == null) { + return result.getResponse(); + } + + // Dynamic step name based on the tool being executed + var toolResult = ctx.step("tool-" + result.getTool().getName(), String.class, + stepCtx -> executeTool(result.getTool(), result.getResponse())); + + messages.add(Map.of("role", "assistant", "content", toolResult)); + } + } +} +``` + ### Human-in-the-Loop Approval **TypeScript:** @@ -187,6 +263,34 @@ def handler(event: dict, context: DurableContext) -> dict: return {'status': 'rejected'} ``` +**Java:** + +```java +import software.amazon.lambda.durable.config.WaitForCallbackConfig; +import software.amazon.lambda.durable.config.CallbackConfig; + +public class ApprovalHandler extends DurableHandler { + @Override + public ApprovalOutput handleRequest(ApprovalInput event, DurableContext ctx) { + var plan = ctx.step("generate-plan", Plan.class, + stepCtx -> generatePlan(event)); + + var answer = ctx.waitForCallback("wait-for-approval", String.class, + (callbackId, stepCtx) -> sendApprovalEmail(event.getApproverEmail(), plan, callbackId), + WaitForCallbackConfig.builder() + .callbackConfig(CallbackConfig.builder().timeout(Duration.ofHours(24)).build()) + .build()); + + if ("APPROVED".equals(answer)) { + ctx.step("execute", Void.class, stepCtx -> { performAction(plan); return null; }); + return new ApprovalOutput("completed"); + } + + return new ApprovalOutput("rejected"); + } +} +``` + ### Saga Pattern (Compensating Transactions) **TypeScript:** @@ -218,6 +322,39 @@ export const handler = withDurableExecution(async (event, context: DurableContex }); ``` +**Java:** + +```java +import software.amazon.lambda.durable.exception.DurableExecutionException; + +public class SagaHandler extends DurableHandler { + @Override + public BookingOutput handleRequest(BookingInput event, DurableContext ctx) { + var compensations = new ArrayList(); + try { + ctx.step("book-flight", Void.class, + s -> { flightClient.book(event); return null; }); + compensations.add(() -> ctx.step("cancel-flight", Void.class, + s -> { flightClient.cancel(event); return null; })); + + ctx.step("book-hotel", Void.class, + s -> { hotelClient.book(event); return null; }); + compensations.add(() -> ctx.step("cancel-hotel", Void.class, + s -> { hotelClient.cancel(event); return null; })); + + return new BookingOutput(true); + } catch (DurableExecutionException e) { + // Run compensations in reverse order + Collections.reverse(compensations); + for (var comp : compensations) { + comp.run(); + } + throw e; + } + } +} +``` + ## Project Structure ### TypeScript @@ -260,6 +397,29 @@ my-durable-function/ └── pyproject.toml # Project configuration ``` +### Java + +``` +my-durable-function/ +├── src/ +│ ├── main/ +│ │ └── java/ +│ │ └── com/example/ +│ │ ├── MyHandler.java # Main handler +│ │ ├── steps/ # Step functions +│ │ │ ├── ValidateStep.java +│ │ │ └── ProcessStep.java +│ │ └── utils/ +│ │ └── RetryStrategies.java +│ └── test/ +│ └── java/ +│ └── com/example/ +│ └── MyHandlerTest.java # Tests with LocalDurableTestRunner +├── infrastructure/ +│ └── template.yaml # SAM/CloudFormation +└── pom.xml # Maven project configuration +``` + ## ESLint Plugin Setup Install the ESLint plugin to catch common durable function mistakes at development time: @@ -344,6 +504,32 @@ module.exports = { Add `aws-durable-execution-sdk-python-testing` to your dev/test dependencies in pyproject.toml. +## Java Maven Setup + +Add the SDK dependencies to your `pom.xml`. Pin to a specific released `VERSION` (see [Maven Central](https://mvnrepository.com/artifact/software.amazon.lambda.durable/aws-durable-execution-sdk-java)) rather than a floating version, and set the Java compiler to 17 or higher: + +```xml + + 17 + 17 + VERSION + + + + + software.amazon.lambda.durable + aws-durable-execution-sdk-java + ${aws-durable-execution-sdk-java.version} + + + software.amazon.lambda.durable + aws-durable-execution-sdk-java-testing + ${aws-durable-execution-sdk-java.version} + test + + +``` + ## Development Workflow ### TypeScript @@ -362,6 +548,14 @@ Add `aws-durable-execution-sdk-python-testing` to your dev/test dependencies in 4. **Deploy** with qualified ARN (version or alias) 5. **Monitor** execution state and logs +### Java + +1. **Write handler** extending `DurableHandler` +2. **Test locally** with `LocalDurableTestRunner` +3. **Validate replay rules** (no non-deterministic code outside steps) +4. **Deploy** with qualified ARN (version or alias) +5. **Monitor** execution state and logs + ## Key Concepts - **Steps**: Atomic operations with automatic retry and checkpointing @@ -397,6 +591,16 @@ When starting a new durable function project: - [ ] Run tests: `pytest` - [ ] Review replay model rules (no non-deterministic code outside steps) +### Java + +- [ ] Add SDK dependencies to `pom.xml` pinned to a specific released `VERSION` +- [ ] Set Java compiler source/target to 17 or higher in `pom.xml` +- [ ] Create handler class extending `DurableHandler` +- [ ] Implement `handleRequest(TInput, DurableContext)` +- [ ] Write tests using `LocalDurableTestRunner` +- [ ] Run tests: `mvn test` +- [ ] Review replay model rules (no non-deterministic code outside steps) + ## Error Scenarios ### Unsupported Language diff --git a/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/replay-model-rules.md b/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/replay-model-rules.md index dc3a6eeb..43b4296b 100644 --- a/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/replay-model-rules.md +++ b/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/replay-model-rules.md @@ -51,6 +51,20 @@ now = datetime.now() # Different datetime each time context.step(lambda _: save_data({"id": id}), name='save') ``` +**Java:** + +```java +// These values change on each replay! +var id = UUID.randomUUID().toString(); // Different UUID each time +var timestamp = System.currentTimeMillis(); // Different timestamp each time +var random = Math.random(); // Different random number +var now = Instant.now(); // Different instant each time + +// The step itself is fine - the problem is that id/timestamp were generated +// OUTSIDE a step, so each replay passes different values into it. +ctx.step("save", Void.class, s -> { saveData(id, timestamp); return null; }); +``` + ### ✅ CORRECT - Non-Deterministic Inside Steps **TypeScript:** @@ -75,6 +89,17 @@ now = context.step(lambda _: datetime.now(), name='get-date') context.step(lambda _: save_data({"id": id}), name='save') ``` +**Java:** + +```java +var id = ctx.step("generate-id", String.class, s -> UUID.randomUUID().toString()); +var timestamp = ctx.step("get-time", Long.class, s -> System.currentTimeMillis()); +var random = ctx.step("random", Double.class, s -> Math.random()); +var now = ctx.step("get-date", Instant.class, s -> Instant.now()); + +ctx.step("save", Void.class, s -> { saveData(id, timestamp); return null; }); +``` + ### Must Be In Steps - `Date.now()`, `new Date()`, `time.time()`, `datetime.now()` @@ -115,6 +140,17 @@ def process(step_ctx: StepContext): context.step(process()) ``` +**Java:** + +```java +ctx.step("process", Result.class, stepCtx -> { + ctx.wait("delay", Duration.ofSeconds(1)); // ERROR! - cannot nest in a step + ctx.step("nested", String.class, s -> ""); // ERROR! - cannot nest in a step + ctx.invoke("other", otherArn, payload, Result.class); // ERROR! - cannot nest in a step + return result; +}); +``` + ### ✅ CORRECT - Use Child Context **TypeScript:** @@ -141,6 +177,17 @@ def process_child(child_ctx: DurableContext): context.run_in_child_context(func=process_child, name='process') ``` +**Java:** + +```java +ctx.runInChildContext("process", Result.class, childCtx -> { + childCtx.wait("processing-delay", Duration.ofSeconds(1)); + var step1 = childCtx.step("validate", Data.class, s -> validate()); + var step2 = childCtx.step("process", Result.class, s -> process(step1)); + return step2; +}); +``` + ## Rule 3: Closure Mutations Are Lost **Variables mutated inside steps are NOT preserved across replays.** @@ -170,6 +217,17 @@ context.step(increment()) print(counter) # Always 0 on replay! ``` +**Java:** + +```java +var counter = new java.util.concurrent.atomic.AtomicInteger(0); +ctx.step("increment", Void.class, s -> { + counter.incrementAndGet(); // This mutation is lost on replay! + return null; +}); +System.out.println(counter.get()); // Always 0 on replay! +``` + ### ✅ CORRECT - Return Values **TypeScript:** @@ -188,6 +246,14 @@ counter = context.step(lambda _: counter + 1, name='increment') print(counter) # Correct value ``` +**Java:** + +```java +int counter = 0; +counter = ctx.step("increment", Integer.class, s -> counter + 1); +System.out.println(counter); // Correct value +``` + ## Rule 4: Side Effects Outside Steps Repeat **Side effects outside steps happen on EVERY replay.** @@ -214,6 +280,16 @@ update_database(data) # Updates multiple times! context.step(lambda _: process(), name='process') ``` +**Java:** + +```java +System.out.println("Starting process"); // Prints multiple times! +sendEmail(user.getEmail()); // Sends multiple emails! +updateDatabase(data); // Updates multiple times! + +ctx.step("process", Result.class, s -> process()); +``` + ### ✅ CORRECT - Side Effects In Steps **TypeScript:** @@ -235,6 +311,15 @@ context.step(update_database(data)) context.step(process()) ``` +**Java:** + +```java +ctx.getLogger().info("Starting process"); // Deduplicated automatically +ctx.step("send-email", Void.class, s -> { sendEmail(user.getEmail()); return null; }); +ctx.step("update-db", Void.class, s -> { updateDatabase(data); return null; }); +ctx.step("process", Result.class, s -> process()); +``` + ### Exception: context.logger `context.logger` is replay-aware and safe to use anywhere. It automatically deduplicates logs across replays. @@ -253,6 +338,16 @@ const apiKey = await context.step('get-key', async () => process.env.API_KEY); await context.step('call-api', async () => callAPI(apiKey)); ``` +```java +// ❌ WRONG if env vars can change +var apiKey = System.getenv("API_KEY"); +ctx.step("call-api", Result.class, s -> callAPI(apiKey)); + +// ✅ CORRECT +var key = ctx.step("get-key", String.class, s -> System.getenv("API_KEY")); +ctx.step("call-api", Result.class, s -> callAPI(key)); +``` + ### Pitfall 2: Array/Object Mutations ```typescript @@ -267,6 +362,22 @@ let items = []; items = await context.step('add-item', async () => [...items, newItem]); ``` +```java +// ❌ WRONG +final var items = new ArrayList(); +ctx.step("add-item", Void.class, s -> { + items.add(newItem); // Lost on replay + return null; +}); + +// ✅ CORRECT +var items = ctx.step("add-item", new TypeToken>() {}, s -> { + var updated = new ArrayList<>(items); + updated.add(newItem); + return updated; +}); +``` + ### Pitfall 3: Conditional Logic with Non-Deterministic Values ```typescript @@ -286,6 +397,23 @@ if (shouldTakePathA) { } ``` +```java +// ❌ WRONG +if (Math.random() > 0.5) { // Different on each replay! + ctx.step("path-a", Result.class, s -> ...); +} else { + ctx.step("path-b", Result.class, s -> ...); +} + +// ✅ CORRECT +var shouldTakePathA = ctx.step("decide", Boolean.class, s -> Math.random() > 0.5); +if (shouldTakePathA) { + ctx.step("path-a", Result.class, s -> ...); +} else { + ctx.step("path-b", Result.class, s -> ...); +} +``` + ## Debugging Replay Issues If you see inconsistent behavior: @@ -308,3 +436,19 @@ const execution = await runner.run({ payload: { test: true } }); const step1 = runner.getOperation('step-name'); expect(step1.getStatus()).toBe(OperationStatus.SUCCEEDED); ``` + +```java +import software.amazon.lambda.durable.testing.LocalDurableTestRunner; +import software.amazon.lambda.durable.model.ExecutionStatus; +import software.amazon.awssdk.services.lambda.model.OperationStatus; + +var runner = LocalDurableTestRunner.create(MyInput.class, new MyHandler()); +// runUntilComplete simulates the Lambda re-invocation (replay) loop +var result = runner.runUntilComplete(new MyInput(true)); + +assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus()); + +// Operation-level status uses OperationStatus +var step1 = result.getOperation("step-name"); +assertEquals(OperationStatus.SUCCEEDED, step1.getStatus()); +``` diff --git a/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/step-operations.md b/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/step-operations.md index 0b3f62db..20a8e561 100644 --- a/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/step-operations.md +++ b/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/step-operations.md @@ -53,6 +53,24 @@ const result = await context.step('fetch-user', async () => { **Best Practice:** Always name steps for easier debugging and testing. +### Java: Typed Steps + +**Java:** + +```java +import software.amazon.lambda.durable.StepContext; + +// Java always requires a name and the result type (Class or TypeToken) +var result = ctx.step("fetch-user", User.class, + (StepContext stepCtx) -> fetchUserFromAPI(userId)); + +// Use TypeToken for generic result types like List +var users = ctx.step("fetch-users", new TypeToken>() {}, + stepCtx -> fetchAllUsers()); +``` + +**Best Practice:** Always name steps for easier debugging and testing. + ## Retry Configuration ### Exponential Backoff @@ -98,6 +116,25 @@ result = context.step( ) ``` +**Java:** + +```java +import java.time.Duration; +import software.amazon.lambda.durable.config.StepConfig; +import software.amazon.lambda.durable.retry.RetryStrategies; +import software.amazon.lambda.durable.retry.JitterStrategy; + +var result = ctx.step("api-call", ApiResponse.class, stepCtx -> callExternalAPI(), + StepConfig.builder() + .retryStrategy(RetryStrategies.exponentialBackoff( + 5, // maxAttempts (including initial attempt) + Duration.ofSeconds(1), // initialDelay + Duration.ofSeconds(60), // maxDelay + 2.0, // backoffRate + JitterStrategy.FULL)) + .build()); +``` + ### Custom Retry Strategy **TypeScript:** @@ -150,6 +187,27 @@ result = context.step( ) ``` +**Java:** + +```java +import software.amazon.lambda.durable.retry.RetryDecision; + +var result = ctx.step("custom-retry", Result.class, stepCtx -> riskyOperation(), + StepConfig.builder() + .retryStrategy((error, attemptCount) -> { + // Don't retry validation errors + if (error instanceof ValidationException) { + return RetryDecision.fail(); + } + // Retry up to 3 times with exponential backoff + if (attemptCount < 3) { + return RetryDecision.retry(Duration.ofSeconds((long) Math.pow(2, attemptCount))); + } + return RetryDecision.fail(); + }) + .build()); +``` + ### Retryable Error Types **TypeScript:** @@ -176,6 +234,26 @@ retry_config = RetryStrategyConfig( ) ``` +**Java:** + +```java +import software.amazon.lambda.durable.retry.RetryDecision; + +// The Java SDK has no "retryable types" option on the built-in strategy; +// inspect the error type inside a custom strategy instead. +var result = ctx.step("selective-retry", Result.class, stepCtx -> operation(), + StepConfig.builder() + .retryStrategy((error, attemptCount) -> { + boolean retryable = error instanceof NetworkException + || error instanceof TimeoutException; + if (retryable && attemptCount < 3) { + return RetryDecision.retry(Duration.ofSeconds((long) Math.pow(2, attemptCount))); + } + return RetryDecision.fail(); + }) + .build()); +``` + ## Step Semantics ### AT_LEAST_ONCE (Default) @@ -217,6 +295,22 @@ result = context.step( ) ``` +**Java:** + +```java +import software.amazon.lambda.durable.config.StepSemantics; + +// AT_LEAST_ONCE_PER_RETRY (default) - may execute multiple times on failure/retry +var result = ctx.step("idempotent-op", Result.class, stepCtx -> idempotentAPI()); + +// AT_MOST_ONCE_PER_RETRY - use for non-idempotent operations. +// semanticsPerRetry(...) retries interrupted steps per the retry strategy. +var payment = ctx.step("charge-payment", PaymentResult.class, stepCtx -> chargeCard(amount), + StepConfig.builder() + .semanticsPerRetry(StepSemantics.AT_MOST_ONCE_PER_RETRY) + .build()); +``` + ## Custom Serialization For complex types, provide custom serialization: @@ -262,6 +356,43 @@ user = context.step( ) ``` +**Java:** + +```java +import software.amazon.lambda.durable.serde.SerDes; +import software.amazon.lambda.durable.TypeToken; +import com.fasterxml.jackson.databind.ObjectMapper; + +// SerDes is a non-generic interface: serialize(Object) and +// deserialize(String, TypeToken) +public class UserSerDes implements SerDes { + private final ObjectMapper mapper = new ObjectMapper(); + + @Override + public String serialize(Object value) { + try { + return mapper.writeValueAsString(value); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + + @Override + public T deserialize(String data, TypeToken typeToken) { + try { + return mapper.readValue(data, mapper.constructType(typeToken.getType())); + } catch (Exception e) { + throw new RuntimeException(e); + } + } +} + +// Provide the custom SerDes via StepConfig +var user = ctx.step("fetch-user", User.class, + stepCtx -> new User("123", "Alice", Instant.now()), + StepConfig.builder().serDes(new UserSerDes()).build()); +``` + ## When to Use Steps vs Child Contexts ### Use Steps For: @@ -295,6 +426,23 @@ await context.runInChildContext('process', async (childCtx) => { }); ``` +**Java:** + +```java +// ❌ WRONG: Cannot nest durable operations in step +ctx.step("process", Result.class, stepCtx -> { + ctx.wait("delay", Duration.ofSeconds(1)); // ERROR! + return result; +}); + +// ✅ CORRECT: Use child context +var result = ctx.runInChildContext("process", ProcessResult.class, childCtx -> { + var data = childCtx.step("fetch", Data.class, s -> fetchData()); + childCtx.wait("processing-delay", Duration.ofSeconds(1)); + return childCtx.step("save", SaveResult.class, s -> save(data)); +}); +``` + ## Error Handling Steps throw errors after all retry attempts are exhausted: @@ -336,6 +484,23 @@ except Exception as error: context.logger.error('Application error: %s', str(error)) ``` +**Java:** + +```java +import software.amazon.lambda.durable.exception.StepFailedException; +import software.amazon.lambda.durable.exception.StepInterruptedException; + +try { + var result = ctx.step("risky", Result.class, stepCtx -> riskyOperation()); +} catch (StepFailedException e) { + ctx.getLogger().error("Step permanently failed: {}", e.getMessage()); + // Handle or rethrow +} catch (StepInterruptedException e) { + ctx.getLogger().error("Step interrupted: {}", e.getMessage()); + // Handle or rethrow +} +``` + ## Best Practices 1. **Always name steps** for debugging and testing diff --git a/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/testing-patterns.md b/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/testing-patterns.md index 21d4f39a..92021cc0 100644 --- a/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/testing-patterns.md +++ b/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/testing-patterns.md @@ -20,6 +20,10 @@ Test durable functions locally and in the cloud with comprehensive test runners. - ✅ Python: Instantiate `DurableFunctionTestRunner(handler=my_handler)` directly - ✅ Python: Use `runner.run(input={...}, timeout=10)` — note `input=` not `payload` - ✅ Python: The value of result.result is serialized. Deserialize using the appropriate SerDes or default json deserializer. +- ✅ Java: Use `LocalDurableTestRunner.create(InputType.class, new MyHandler())` (first arg is the handler INPUT type) +- ✅ Java: Use `runner.runUntilComplete(input)` to drive the replay loop; `run(input)` for a single invocation +- ✅ Java: `result.getStatus()` returns `ExecutionStatus`; `operation.getStatus()` returns `OperationStatus` +- ✅ Java: Resolve callbacks with `runner.getCallbackId("-callback")` then `runner.completeCallback(id, jsonResult)` / `runner.failCallback(id, error)` (waitForCallback appends `-callback` to the operation name) ### DON'T: @@ -30,6 +34,7 @@ Test durable functions locally and in the cloud with comprehensive test runners. - ❌ TypeScript: Test callbacks without proper synchronization (leads to race conditions) - ❌ Python: Confuse `DurableFunctionTestRunner` (local) with `DurableFunctionCloudTestRunner` (cloud) - ❌ Python: Forget the `with runner:` context manager — it manages execution lifecycle +- ❌ Java: Assert `result.getStatus()` against `OperationStatus` — it returns `ExecutionStatus` (use `OperationStatus` only for individual operations) ## Local Testing Setup @@ -93,6 +98,30 @@ def test_workflow(): assert result.status is InvocationStatus.SUCCEEDED ``` +**Java:** + +The Java testing SDK provides `LocalDurableTestRunner` for local testing and `CloudDurableTestRunner` for cloud testing. + +```java +import software.amazon.lambda.durable.testing.LocalDurableTestRunner; +import software.amazon.lambda.durable.model.ExecutionStatus; +import org.junit.jupiter.api.Test; +import static org.junit.jupiter.api.Assertions.*; + +@Test +void shouldExecuteWorkflow() { + // First arg is the handler INPUT type + var runner = LocalDurableTestRunner.create(MyInput.class, new MyHandler()); + + // runUntilComplete simulates the Lambda re-invocation loop (advances waits/retries) + var result = runner.runUntilComplete(new MyInput("123")); + + // result.getStatus() returns ExecutionStatus + assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus()); + assertEquals(new MyOutput(true), result.getResult(MyOutput.class)); +} +``` + ## Getting Operations **CRITICAL: Always get operations by NAME, not by index.** @@ -138,6 +167,29 @@ def test_steps_execute(): assert 'process-data' in step_names ``` +**Java:** + +```java +import software.amazon.awssdk.services.lambda.model.OperationStatus; +import software.amazon.awssdk.services.lambda.model.OperationType; + +@Test +void shouldExecuteStepsInOrder() { + var runner = LocalDurableTestRunner.create(MyInput.class, new MyHandler()); + var result = runner.runUntilComplete(new MyInput(true)); + + // ✅ CORRECT: Get by name (operation status uses OperationStatus) + var fetchStep = result.getOperation("fetch-user"); + assertEquals(OperationType.STEP, fetchStep.getType()); + assertEquals(OperationStatus.SUCCEEDED, fetchStep.getStatus()); + + var processStep = result.getOperation("process-data"); + assertEquals(OperationStatus.SUCCEEDED, processStep.getStatus()); + + // ❌ WRONG: relying on positional/index lookup is brittle +} +``` + ## Testing Replay Behavior **TypeScript:** @@ -159,6 +211,25 @@ it('should handle replay correctly', async () => { }); ``` +**Java:** + +```java +@Test +void shouldHandleReplayCorrectly() { + var runner = LocalDurableTestRunner.create(MyInput.class, new MyHandler()); + + // runUntilComplete drives the function through its replay cycles + var result = runner.runUntilComplete(new MyInput(42)); + assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus()); + + // A fresh runner replays from the start and must produce the same result + var replayRunner = LocalDurableTestRunner.create(MyInput.class, new MyHandler()); + var replayResult = replayRunner.runUntilComplete(new MyInput(42)); + assertEquals(ExecutionStatus.SUCCEEDED, replayResult.getStatus()); + assertEquals(result.getResult(MyOutput.class), replayResult.getResult(MyOutput.class)); +} +``` + ## Testing with Fake Clock **TypeScript:** @@ -183,6 +254,24 @@ it('should wait for specified duration', async () => { }); ``` +**Java:** + +```java +@Test +void shouldWaitForSpecifiedDuration() { + var runner = LocalDurableTestRunner.create(MyInput.class, new MyHandler()); + + // runUntilComplete automatically advances WAIT operations (no real time elapses) + var result = runner.runUntilComplete(new MyInput()); + + assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus()); + + var waitOp = result.getOperation("delay"); + assertEquals(OperationType.WAIT, waitOp.getType()); + assertEquals(OperationStatus.SUCCEEDED, waitOp.getStatus()); +} +``` + ## Test Runner API Patterns **CRITICAL:** Always wrap event data in `payload` and cast results appropriately. @@ -226,6 +315,26 @@ it('incorrect api usage', async () => { }); ``` +**Java:** + +```java +@Test +void shouldUseCorrectTestRunnerApi() { + var runner = LocalDurableTestRunner.create(GreetingInput.class, new GreetingHandler()); + + // Typed input - no payload wrapper needed in Java + var result = runner.runUntilComplete(new GreetingInput("Alice", "123")); + + // Typed result via getResult(Class) + var output = result.getResult(GreetingOutput.class); + assertEquals("Hello, Alice!", output.getGreeting()); + + // Get operations by name; read the typed step result + var greetingStep = result.getOperation("generate-greeting"); + assertEquals("Hello, Alice!", greetingStep.getStepResult(String.class)); +} +``` + ## Testing Callbacks **CRITICAL:** Use `waitForData()` with `WaitingOperationStatus.STARTED` to avoid flaky tests caused by promise races. @@ -310,6 +419,48 @@ def test_callback_creation(): assert callback_ops[0].callback_id is not None ``` +**Java:** + +```java +import software.amazon.awssdk.services.lambda.model.ErrorObject; + +@Test +void shouldHandleCallbackSuccess() { + var runner = LocalDurableTestRunner.create(ApprovalInput.class, new ApprovalHandler()); + + // First invocation suspends at the callback (execution status PENDING) + var pending = runner.run(new ApprovalInput("alice@example.com")); + assertEquals(ExecutionStatus.PENDING, pending.getStatus()); + + // Resolve the callback by operation name; the result payload is JSON. + // waitForCallback("wait-for-approval", ...) creates a nested CALLBACK operation + // named "wait-for-approval-callback" - resolve by that name. + var callbackId = runner.getCallbackId("wait-for-approval-callback"); + runner.completeCallback(callbackId, "{\"approved\":true,\"comments\":\"Looks good\"}"); + + // Resume to completion + var result = runner.runUntilComplete(new ApprovalInput("alice@example.com")); + assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus()); +} + +@Test +void shouldHandleCallbackFailure() { + var runner = LocalDurableTestRunner.create(ApprovalInput.class, new ApprovalHandler()); + + var pending = runner.run(new ApprovalInput("alice@example.com")); + assertEquals(ExecutionStatus.PENDING, pending.getStatus()); + + var callbackId = runner.getCallbackId("wait-for-approval-callback"); + runner.failCallback(callbackId, ErrorObject.builder() + .errorType("ApprovalDenied") + .errorMessage("Request was rejected") + .build()); + + var result = runner.runUntilComplete(new ApprovalInput("alice@example.com")); + assertEquals(ExecutionStatus.FAILED, result.getStatus()); +} +``` + ## Testing Callback Heartbeats **TypeScript:** @@ -338,6 +489,26 @@ it('should handle callback heartbeats', async () => { }); ``` +**Java:** + +```java +@Test +void shouldHandleLongRunningCallback() { + var runner = LocalDurableTestRunner.create(MyInput.class, new MyHandler()); + + // Suspends at the long-running callback + var pending = runner.run(new MyInput()); + assertEquals(ExecutionStatus.PENDING, pending.getStatus()); + + // Complete the callback once the external process finishes + var callbackId = runner.getCallbackId("long-running-process-callback"); + runner.completeCallback(callbackId, "{\"status\":\"completed\"}"); + + var result = runner.runUntilComplete(new MyInput()); + assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus()); +} +``` + ## Testing Error Scenarios **TypeScript:** @@ -387,6 +558,31 @@ it('should fail after max retries', async () => { }); ``` +**Java:** + +```java +@Test +void shouldRetryOnFailure() { + var runner = LocalDurableTestRunner.create(MyInput.class, new RetryHandler()); + var result = runner.runUntilComplete(new MyInput()); + + assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus()); + + var step = result.getOperation("flaky-operation"); + assertEquals(OperationStatus.SUCCEEDED, step.getStatus()); +} + +@Test +void shouldFailAfterMaxRetries() { + var runner = LocalDurableTestRunner.create(MyInput.class, new AlwaysFailHandler()); + var result = runner.runUntilComplete(new MyInput()); + + assertEquals(ExecutionStatus.FAILED, result.getStatus()); + assertTrue(result.getError().isPresent()); + assertTrue(result.getError().get().errorMessage().contains("Permanent failure")); +} +``` + ## Testing Concurrent Operations **TypeScript:** @@ -410,6 +606,25 @@ it('should process items concurrently', async () => { }); ``` +**Java:** + +```java +@Test +void shouldProcessItemsConcurrently() { + var runner = LocalDurableTestRunner.create(MyInput.class, new ConcurrentHandler()); + + var result = runner.runUntilComplete(new MyInput(List.of(1, 2, 3, 4, 5))); + assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus()); + + var mapOp = result.getOperation("process-items"); + assertEquals(OperationType.MAP, mapOp.getType()); + + // Check an individual item operation (map items run in named child contexts) + var item0 = result.getOperation("process-0"); + assertEquals(OperationStatus.SUCCEEDED, item0.getStatus()); +} +``` + ## Cloud Testing For integration tests against real Lambda: @@ -469,6 +684,28 @@ def test_workflow_cloud(): assert result.status is InvocationStatus.SUCCEEDED ``` +**Java:** + +```java +import software.amazon.lambda.durable.testing.CloudDurableTestRunner; + +@Test +void shouldExecuteInRealLambda() { + // create(functionArn, inputType, outputType); region/credentials come from + // the default AWS provider chain (or supply a custom LambdaClient) + var runner = CloudDurableTestRunner.create( + "my-durable-function:1", // Qualified ARN required + MyInput.class, + MyOutput.class); + + var result = runner.runUntilComplete(new MyInput("123")); + assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus()); + + var step = result.getOperation("fetch-user"); + assertEquals(OperationStatus.SUCCEEDED, step.getStatus()); +} +``` + ## Test Assertions **TypeScript:** @@ -496,6 +733,28 @@ it('should validate operation details', async () => { }); ``` +**Java:** + +```java +@Test +void shouldValidateOperationDetails() { + var runner = LocalDurableTestRunner.create(MyInput.class, new MyHandler()); + var result = runner.runUntilComplete(new MyInput()); + + var step = result.getOperation("process-data"); + + // Operation type and status + assertEquals(OperationType.STEP, step.getType()); + assertEquals(OperationStatus.SUCCEEDED, step.getStatus()); + + // Timing + assertNotNull(step.getDuration()); + + // Typed step result + assertEquals(new ProcessResult(true), step.getStepResult(ProcessResult.class)); +} +``` + ## Best Practices 1. **Always name operations** for reliable test assertions @@ -558,6 +817,16 @@ await callbackOp.sendCallbackSuccess(JSON.stringify({})); | Callback result parsing error | Result is JSON string | Parse result: `JSON.parse(result.value)` | | Operation not found by name | Missing operation name | Always name operations in handler | +### Java + +| Error | Cause | Solution | +| ------------------------------------------------ | ---------------------------------------------- | --------------------------------------------------------------- | +| `incompatible types: ExecutionStatus vs OperationStatus` | Comparing `result.getStatus()` to `OperationStatus` | Use `ExecutionStatus` for the execution; `OperationStatus` for operations | +| `ResultType cannot be null` | Calling no-arg `getResult()` without an output type | Use `result.getResult(MyOutput.class)` | +| Callback never resolves / stays PENDING | Forgot to resolve the callback, or used the wrong name | `waitForCallback("x", ...)` creates a callback op named `x-callback`; call `runner.getCallbackId("x-callback")` then `completeCallback(...)` | +| Wrong constructor args | Passing output type first to `create(...)` | First arg is the INPUT type: `create(MyInput.class, handler)` | +| Operation not found by name | Missing operation name | Always name operations in the handler | + ## Jest Configuration **jest.config.js:** diff --git a/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/wait-operations.md b/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/wait-operations.md index 664c1925..cc2621fb 100644 --- a/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/wait-operations.md +++ b/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/wait-operations.md @@ -32,6 +32,23 @@ context.wait(duration=Duration.from_days(7)) context.wait(duration=Duration.from_seconds(60), name='rate-limit-delay') ``` +**Java:** + +```java +import java.time.Duration; + +// Synchronous wait (blocks execution) +ctx.wait("delay", Duration.ofSeconds(30)); +ctx.wait("rate-limit-delay", Duration.ofMinutes(5)); +ctx.wait("long-delay", Duration.ofHours(1)); +ctx.wait("very-long-delay", Duration.ofDays(7)); + +// Async wait (returns DurableFuture) +var future = ctx.waitAsync("async-delay", Duration.ofHours(1)); +// ... do other work ... +future.get(); +``` + **Max wait duration:** Up to 1 year ## Wait for Callback @@ -79,6 +96,29 @@ result = context.wait_for_callback( ) ``` +**Java:** + +```java +import software.amazon.lambda.durable.config.WaitForCallbackConfig; +import software.amazon.lambda.durable.config.CallbackConfig; + +var result = ctx.waitForCallback("wait-for-approval", ApprovalResult.class, + (callbackId, stepCtx) -> { + sendApprovalEmail(approverEmail, callbackId); + }, + WaitForCallbackConfig.builder() + .callbackConfig(CallbackConfig.builder() + .timeout(Duration.ofHours(24)) + .heartbeatTimeout(Duration.ofMinutes(5)) + .build()) + .build()); + +// External system calls back with: +// aws lambda send-durable-execution-callback-success \ +// --callback-id \ +// --payload '{"approved": true}' +``` + ### Callback Success **CLI:** @@ -114,6 +154,21 @@ lambda_client.send_durable_execution_callback_success( ) ``` +**SDK (Java):** + +```java +import software.amazon.awssdk.services.lambda.LambdaClient; +import software.amazon.awssdk.services.lambda.model.SendDurableExecutionCallbackSuccessRequest; + +LambdaClient client = LambdaClient.create(); +client.sendDurableExecutionCallbackSuccess( + SendDurableExecutionCallbackSuccessRequest.builder() + .callbackId(callbackId) + .payload("{\"status\":\"approved\"}") + .build() +); +``` + ### Callback Failure **CLI:** @@ -212,6 +267,41 @@ result = context.wait_for_condition( ) ``` +**Java:** + +```java +import software.amazon.lambda.durable.config.WaitForConditionConfig; +import software.amazon.lambda.durable.model.WaitForConditionResult; +import software.amazon.lambda.durable.exception.WaitForConditionFailedException; + +// Define state class for job polling +record JobState(String jobId, String status) {} + +var finalState = ctx.waitForCondition("wait-for-job", JobState.class, + (currentState, stepCtx) -> { + // Check current status and update state + var status = checkJobStatus(currentState.jobId()); + var newState = new JobState(currentState.jobId(), status); + + // Decision logic: stop if completed, continue otherwise + if ("completed".equals(status)) { + return WaitForConditionResult.stopPolling(newState); + } + return WaitForConditionResult.continuePolling(newState); + }, + WaitForConditionConfig.builder() + .initialState(new JobState("job-123", "pending")) + .waitStrategy((state, attempt) -> { + // Compute delay duration (not decision logic) + if (attempt >= 60) { + throw new WaitForConditionFailedException("Max polling attempts exceeded"); + } + long delaySeconds = Math.min((long) (5 * Math.pow(1.5, attempt)), 30); + return Duration.ofSeconds(delaySeconds); + }) + .build()); +``` + ### Custom Wait Strategy **TypeScript:** @@ -241,6 +331,38 @@ const result = await context.waitForCondition( ); ``` +**Java:** + +```java +import software.amazon.lambda.durable.model.WaitForConditionResult; + +// Define state class +record PollState(Data data, int attempts) { + boolean isReady() { return data != null && data.isReady(); } +} + +var result = ctx.waitForCondition("custom-poll", PollState.class, + (state, stepCtx) -> { + // Fetch data and update state + var data = fetchData(); + var newState = new PollState(data, state.attempts() + 1); + + // Decision logic: stop if ready or max attempts + if (newState.isReady() || newState.attempts() >= 10) { + return WaitForConditionResult.stopPolling(newState); + } + return WaitForConditionResult.continuePolling(newState); + }, + WaitForConditionConfig.builder() + .initialState(new PollState(null, 0)) + .waitStrategy((state, attempt) -> { + // Compute exponential backoff delay (max 60 seconds) + long delaySeconds = Math.min((long) Math.pow(2, attempt), 60); + return Duration.ofSeconds(delaySeconds); + }) + .build()); +``` + ## Callback Patterns ### Human Approval Workflow @@ -275,6 +397,41 @@ export const handler = withDurableExecution(async (event, context: DurableContex }); ``` +**Java:** + +```java +import software.amazon.lambda.durable.config.WaitForCallbackConfig; +import software.amazon.lambda.durable.config.CallbackConfig; + +public class ApprovalHandler extends DurableHandler { + @Override + public ApprovalResult handleRequest(ApprovalRequest event, DurableContext ctx) { + var request = ctx.step("create-request", Request.class, + s -> createApprovalRequest(event)); + + var decision = ctx.waitForCallback("wait-approval", Decision.class, + (callbackId, s) -> { + sendEmail(Map.of( + "to", event.getApprover(), + "subject", "Approval Required", + "body", "Approve: " + approvalUrl + "?callback=" + callbackId + "&action=approve\n" + + "Reject: " + approvalUrl + "?callback=" + callbackId + "&action=reject" + )); + }, + WaitForCallbackConfig.builder() + .callbackConfig(CallbackConfig.builder().timeout(Duration.ofHours(48)).build()) + .build()); + + if ("approve".equals(decision.getAction())) { + ctx.step("execute", Void.class, s -> { executeRequest(request); return null; }); + return new ApprovalResult("approved"); + } + + return new ApprovalResult("rejected"); + } +} +``` + ### Webhook Integration **TypeScript:** @@ -305,6 +462,39 @@ export const handler = withDurableExecution(async (event, context: DurableContex }); ``` +**Java:** + +```java +import software.amazon.lambda.durable.config.WaitForCallbackConfig; +import software.amazon.lambda.durable.config.CallbackConfig; + +public class OrderHandler extends DurableHandler { + @Override + public OrderResult handleRequest(OrderRequest event, DurableContext ctx) { + var order = ctx.step("create-order", Order.class, + s -> createOrder(event)); + + var payment = ctx.waitForCallback("wait-payment", Payment.class, + (callbackId, s) -> { + paymentProvider.createPayment(Map.of( + "orderId", order.getId(), + "amount", order.getTotal(), + "webhookUrl", webhookUrl + "?callback=" + callbackId + )); + }, + WaitForCallbackConfig.builder() + .callbackConfig(CallbackConfig.builder().timeout(Duration.ofMinutes(15)).build()) + .build()); + + if ("success".equals(payment.getStatus())) { + ctx.step("fulfill", Void.class, s -> { fulfillOrder(order); return null; }); + } + + return new OrderResult(order.getId(), payment.getStatus()); + } +} +``` + ### Async Job Polling **TypeScript:** @@ -338,6 +528,50 @@ export const handler = withDurableExecution(async (event, context: DurableContex }); ``` +**Java:** + +```java +import software.amazon.lambda.durable.model.WaitForConditionResult; +import software.amazon.lambda.durable.exception.WaitForConditionFailedException; + +// Define state class for job polling +record JobState(String jobId, String status, String result) {} + +public class JobHandler extends DurableHandler { + @Override + public JobResult handleRequest(JobRequest event, DurableContext ctx) { + var jobId = ctx.step("start-job", String.class, + s -> startBatchJob(event.getData())); + + var result = ctx.waitForCondition("poll-job", JobState.class, + (state, stepCtx) -> { + // Fetch current job status + var job = getJobStatus(state.jobId()); + var newState = new JobState(state.jobId(), job.getStatus(), job.getResult()); + + // Decision logic: stop if not running, continue otherwise + if (!"running".equals(newState.status())) { + return WaitForConditionResult.stopPolling(newState); + } + return WaitForConditionResult.continuePolling(newState); + }, + WaitForConditionConfig.builder() + .initialState(new JobState(jobId, "running", null)) + .waitStrategy((state, attempt) -> { + // Compute delay with exponential backoff + if (attempt >= 60) { + throw new WaitForConditionFailedException("Max polling attempts exceeded"); + } + long delaySeconds = Math.min((long) (5 * Math.pow(1.5, attempt)), 30); + return Duration.ofSeconds(delaySeconds); + }) + .build()); + + return new JobResult(result.jobId(), result.status(), result.result()); + } +} +``` + ## Best Practices 1. **Always name wait operations** for debugging @@ -394,3 +628,30 @@ except CallbackError as error: else: context.logger.error('Callback failed', error) ``` + +**Java:** + +```java +import software.amazon.lambda.durable.config.WaitForCallbackConfig; +import software.amazon.lambda.durable.config.CallbackConfig; +import software.amazon.lambda.durable.exception.CallbackFailedException; +import software.amazon.lambda.durable.exception.CallbackTimeoutException; +import software.amazon.lambda.durable.exception.WaitForConditionFailedException; + +try { + var result = ctx.waitForCallback("wait-approval", ApprovalResult.class, + (callbackId, stepCtx) -> sendApproval(callbackId), + WaitForCallbackConfig.builder() + .callbackConfig(CallbackConfig.builder().timeout(Duration.ofHours(24)).build()) + .build()); +} catch (CallbackTimeoutException e) { + ctx.getLogger().warn("Approval timed out: {}", e.getMessage()); + // Handle timeout +} catch (CallbackFailedException e) { + ctx.getLogger().error("Callback failed: {}", e.getMessage()); + // Handle failure +} catch (WaitForConditionFailedException e) { + ctx.getLogger().error("Condition polling failed: {}", e.getMessage()); + // Handle polling failure +} +``` From 6876320cce7894eda598ba9ed841d72ac4aa3bdb Mon Sep 17 00:00:00 2001 From: DeepaliTandale <18361446+DeepaliTandale@users.noreply.github.com> Date: Mon, 15 Jun 2026 15:43:25 +0100 Subject: [PATCH 2/5] added map operation as child context in testing-patterns.md --- .../references/testing-patterns.md | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/testing-patterns.md b/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/testing-patterns.md index 92021cc0..2e4b47a3 100644 --- a/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/testing-patterns.md +++ b/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/testing-patterns.md @@ -616,10 +616,12 @@ void shouldProcessItemsConcurrently() { var result = runner.runUntilComplete(new MyInput(List.of(1, 2, 3, 4, 5))); assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus()); - var mapOp = result.getOperation("process-items"); - assertEquals(OperationType.MAP, mapOp.getType()); + // The map operation runs each item in its own named child CONTEXT. + // (OperationType has EXECUTION, CONTEXT, STEP, WAIT, CALLBACK, CHAINED_INVOKE - there is no MAP.) + var mapContext = result.getOperation("process-items"); + assertEquals(OperationType.CONTEXT, mapContext.getType()); - // Check an individual item operation (map items run in named child contexts) + // Check an individual item's step operation inside the map var item0 = result.getOperation("process-0"); assertEquals(OperationStatus.SUCCEEDED, item0.getStatus()); } From 69641045ed0f7dc8b8d4a12db9d54c240c26ee1b Mon Sep 17 00:00:00 2001 From: DeepaliTandale <18361446+DeepaliTandale@users.noreply.github.com> Date: Mon, 15 Jun 2026 16:12:21 +0100 Subject: [PATCH 3/5] replaced callback payload(string) with result(sdkbytes) as per SDK doc --- .../aws-lambda-durable-functions/references/wait-operations.md | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/wait-operations.md b/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/wait-operations.md index cc2621fb..283042f8 100644 --- a/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/wait-operations.md +++ b/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/wait-operations.md @@ -157,6 +157,7 @@ lambda_client.send_durable_execution_callback_success( **SDK (Java):** ```java +import software.amazon.awssdk.core.SdkBytes; import software.amazon.awssdk.services.lambda.LambdaClient; import software.amazon.awssdk.services.lambda.model.SendDurableExecutionCallbackSuccessRequest; @@ -164,7 +165,7 @@ LambdaClient client = LambdaClient.create(); client.sendDurableExecutionCallbackSuccess( SendDurableExecutionCallbackSuccessRequest.builder() .callbackId(callbackId) - .payload("{\"status\":\"approved\"}") + .result(SdkBytes.fromUtf8String("{\"status\":\"approved\"}")) .build() ); ``` From 390591b07b2bab89015e2aa20352b44fae889bb9 Mon Sep 17 00:00:00 2001 From: DeepaliTandale <18361446+DeepaliTandale@users.noreply.github.com> Date: Mon, 15 Jun 2026 17:24:48 +0100 Subject: [PATCH 4/5] added java virtual threads and checkpoint dealy config --- .../references/advanced-patterns.md | 59 +++++++++++++++++++ 1 file changed, 59 insertions(+) diff --git a/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/advanced-patterns.md b/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/advanced-patterns.md index 3ae66998..e6a91fd7 100644 --- a/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/advanced-patterns.md +++ b/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/advanced-patterns.md @@ -490,6 +490,65 @@ var result = ctx.step("process-order", Order.class, stepCtx -> { }); ``` +## Java SDK Configuration + +The Java SDK is configured per-handler by overriding `createConfiguration()` on +`DurableHandler` and returning a `DurableConfig`. This is the Java-specific entry point for +tuning serialization, the concurrency executor, and checkpoint batching. (TypeScript and +Python expose equivalent configuration through their own SDK options.) + +> **Note:** The virtual-thread executor below requires **Java 21+** +> (`Executors.newVirtualThreadPerTaskExecutor()`). The SDK itself targets Java 17+; on +> Java 17 use a platform-thread pool such as `Executors.newFixedThreadPool(n)` instead. + +**Java:** + +```java +import java.time.Duration; +import java.util.concurrent.Executors; +import software.amazon.lambda.durable.DurableConfig; +import software.amazon.lambda.durable.DurableContext; +import software.amazon.lambda.durable.DurableFuture; +import software.amazon.lambda.durable.DurableHandler; + +public class ManyStepsHandler extends DurableHandler { + + @Override + protected DurableConfig createConfiguration() { + return DurableConfig.builder() + // Virtual threads scale to large numbers of concurrent async operations + // (Java 21+). Recommended when a handler fans out into many + // stepAsync / map / parallel branches. + .withExecutorService(Executors.newVirtualThreadPerTaskExecutor()) + // A small checkpoint delay batches checkpoint requests together, reducing + // overall latency when there are many concurrent operations. + .withCheckpointDelay(Duration.ofMillis(10)) + .build(); + } + + @Override + public Integer handleRequest(BatchInput event, DurableContext ctx) { + // Fan out into many async steps - executed on the virtual-thread pool above + var futures = new java.util.ArrayList>(); + for (int i = 0; i < event.getCount(); i++) { + int index = i; + futures.add(ctx.stepAsync("compute-" + i, Integer.class, s -> index * 2)); + } + + // Collect all results in order + var results = DurableFuture.allOf(futures); + return results.stream().mapToInt(Integer::intValue).sum(); + } +} +``` + +**When to use these settings:** + +| Setting | Use when | +| --- | --- | +| `withExecutorService(Executors.newVirtualThreadPerTaskExecutor())` | The handler creates many concurrent operations (`stepAsync`, `map`, `parallel`); virtual threads (Java 21+) scale far better than a fixed thread pool | +| `withCheckpointDelay(Duration.ofMillis(...))` | Many concurrent operations checkpoint at once; a small delay batches the checkpoint API calls and lowers latency | + ## Nested Workflows ### Parent-Child Workflow Pattern From 821575032fdab98d7ffd26ce9d71a8ad0b9cfe97 Mon Sep 17 00:00:00 2001 From: DeepaliTandale <18361446+DeepaliTandale@users.noreply.github.com> Date: Wed, 8 Jul 2026 11:42:46 +0100 Subject: [PATCH 5/5] fix(aws-serverless): address 3 reviewer comments on durable-functions Java - replay-model-rules.md Rule 3: fix effectively-final lambda capture (use separate initial var) - replay-model-rules.md: remove fully-qualified AtomicInteger, use short class name - SKILL.md: fix return type mismatch (ProcessResult -> MyOutput) in quick-reference handler --- .../skills/aws-lambda-durable-functions/SKILL.md | 2 +- .../references/replay-model-rules.md | 6 +++--- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/plugins/aws-serverless/skills/aws-lambda-durable-functions/SKILL.md b/plugins/aws-serverless/skills/aws-lambda-durable-functions/SKILL.md index 4ba1519a..61bc507e 100644 --- a/plugins/aws-serverless/skills/aws-lambda-durable-functions/SKILL.md +++ b/plugins/aws-serverless/skills/aws-lambda-durable-functions/SKILL.md @@ -161,7 +161,7 @@ import software.amazon.lambda.durable.DurableContext; public class MyHandler extends DurableHandler { @Override public MyOutput handleRequest(MyInput event, DurableContext ctx) { - var result = ctx.step("process", ProcessResult.class, stepCtx -> processData(event)); + var result = ctx.step("process", MyOutput.class, stepCtx -> processData(event)); return result; } } diff --git a/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/replay-model-rules.md b/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/replay-model-rules.md index 43b4296b..fc58af6b 100644 --- a/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/replay-model-rules.md +++ b/plugins/aws-serverless/skills/aws-lambda-durable-functions/references/replay-model-rules.md @@ -220,7 +220,7 @@ print(counter) # Always 0 on replay! **Java:** ```java -var counter = new java.util.concurrent.atomic.AtomicInteger(0); +var counter = new AtomicInteger(0); ctx.step("increment", Void.class, s -> { counter.incrementAndGet(); // This mutation is lost on replay! return null; @@ -249,8 +249,8 @@ print(counter) # Correct value **Java:** ```java -int counter = 0; -counter = ctx.step("increment", Integer.class, s -> counter + 1); +int initial = 0; +int counter = ctx.step("increment", Integer.class, s -> initial + 1); System.out.println(counter); // Correct value ```