diff --git a/conformance-tests/src/main/java/map/MapItemNamer.java b/conformance-tests/src/main/java/map/MapItemNamer.java new file mode 100644 index 000000000..5f0b122d5 --- /dev/null +++ b/conformance-tests/src/main/java/map/MapItemNamer.java @@ -0,0 +1,24 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package map; + +import java.util.List; +import software.amazon.lambda.durable.DurableContext; +import software.amazon.lambda.durable.DurableHandler; +import software.amazon.lambda.durable.config.MapConfig; +import software.amazon.lambda.durable.model.MapResult; + +/** 9-13: Map with a custom item namer (names each iteration from its item). */ +public class MapItemNamer extends DurableHandler> { + + @Override + public List handleRequest(Object input, DurableContext context) { + var config = MapConfig.builder() + .maxConcurrency(1) + .itemNamer((item, index) -> "item-" + item) + .build(); + MapResult result = + context.map("named-items", List.of(1, 2), Integer.class, (item, index, ctx) -> item * 10, config); + return result.results(); + } +} diff --git a/conformance-tests/src/main/java/plugin/PluginFaultyAndHealthy.java b/conformance-tests/src/main/java/plugin/PluginFaultyAndHealthy.java index 744b6aa1f..34ca77378 100644 --- a/conformance-tests/src/main/java/plugin/PluginFaultyAndHealthy.java +++ b/conformance-tests/src/main/java/plugin/PluginFaultyAndHealthy.java @@ -18,9 +18,9 @@ * *

A single greeting step configured with TWO plugins registered together, in order: first a faulty plugin whose * every exercised hook (invocation-start, operation-start, attempt-start, attempt-end, operation-end, invocation-end) - * logs a record then throws, then a healthy plugin that logs the corresponding six records normally. The SDK's {@code - * PluginRunner} isolates each plugin at every hook boundary (swallows the faulty plugin's exceptions), so the healthy - * plugin still receives every hook and the execution result/history are identical to running without the faulty + * logs a record then throws, then a healthy plugin that logs the corresponding six records normally. The SDK's + * {@code PluginRunner} isolates each plugin at every hook boundary (swallows the faulty plugin's exceptions), so the + * healthy plugin still receives every hook and the execution result/history are identical to running without the faulty * plugin. Attempt boundaries are the real user-function hooks ({@code onUserFunctionStart}/{@code onUserFunctionEnd}, * filtered to step attempts); the healthy attempt-end reports the SDK's real success/failure outcome. */ diff --git a/conformance-tests/src/main/java/plugin/PluginParallelBranchHooks.java b/conformance-tests/src/main/java/plugin/PluginParallelBranchHooks.java index f4292c447..4931c2e3b 100644 --- a/conformance-tests/src/main/java/plugin/PluginParallelBranchHooks.java +++ b/conformance-tests/src/main/java/plugin/PluginParallelBranchHooks.java @@ -20,8 +20,8 @@ * *

A parallel operation named "parallel" with two branches (max-concurrency 1, so they run sequentially in index * order); each branch returns a constant directly. The plugin, filtering to parallel-branch operations, logs fn-start - * and fn-end (with outcome) from the real user-function hooks, carrying the branch operation id and the parallel - * parent id. These hooks run on the branch's own thread, so start-before-end order per branch is deterministic. + * and fn-end (with outcome) from the real user-function hooks, carrying the branch operation id and the parallel parent + * id. These hooks run on the branch's own thread, so start-before-end order per branch is deterministic. */ @SuppressWarnings("deprecation") public class PluginParallelBranchHooks extends DurableHandler> { diff --git a/conformance-tests/src/main/java/plugin/PluginReplayFlags.java b/conformance-tests/src/main/java/plugin/PluginReplayFlags.java index 0a061f76a..73e8b49fd 100644 --- a/conformance-tests/src/main/java/plugin/PluginReplayFlags.java +++ b/conformance-tests/src/main/java/plugin/PluginReplayFlags.java @@ -17,10 +17,10 @@ /** * 10-13: Non-terminal operations replay with replay=true; terminal operations are not re-emitted. * - *

Two sequential steps. Step A succeeds on its first attempt (terminal before the retry invocation). Step B fails - * on its first attempt and succeeds on the second, using the SDK's built-in exponential-backoff retry strategy - * (~1s delay). The plugin, filtering to step-type operations, logs operation-start with the SDK's is-replayed - * indicator ({@code OperationInfo#isReplay()}) and operation-end with the terminal status. + *

Two sequential steps. Step A succeeds on its first attempt (terminal before the retry invocation). Step B fails on + * its first attempt and succeeds on the second, using the SDK's built-in exponential-backoff retry strategy (~1s + * delay). The plugin, filtering to step-type operations, logs operation-start with the SDK's is-replayed indicator + * ({@code OperationInfo#isReplay()}) and operation-end with the terminal status. */ @SuppressWarnings("deprecation") public class PluginReplayFlags extends DurableHandler { diff --git a/conformance-tests/src/main/java/plugin/PluginRetryExhaustion.java b/conformance-tests/src/main/java/plugin/PluginRetryExhaustion.java index 642f99a03..782d22474 100644 --- a/conformance-tests/src/main/java/plugin/PluginRetryExhaustion.java +++ b/conformance-tests/src/main/java/plugin/PluginRetryExhaustion.java @@ -18,8 +18,8 @@ /** * 10-15: Attempt hooks fire for every attempt until exhaustion, then operation-end reports FAILED. * - *

A single step that always throws, configured with the SDK's built-in exponential-backoff retry strategy allowing - * 2 total attempts (1 initial + 1 retry, ~1s delay). The plugin, filtering to step-type operations, logs attempt-start + *

A single step that always throws, configured with the SDK's built-in exponential-backoff retry strategy allowing 2 + * total attempts (1 initial + 1 retry, ~1s delay). The plugin, filtering to step-type operations, logs attempt-start * and attempt-end (with outcome) from the real user-function hooks (which carry the 1-based attempt number) and * operation-end when the step reaches its terminal FAILED status. */ diff --git a/conformance-tests/src/main/java/plugin/PluginSupport.java b/conformance-tests/src/main/java/plugin/PluginSupport.java index 7260a0c43..474f94e6c 100644 --- a/conformance-tests/src/main/java/plugin/PluginSupport.java +++ b/conformance-tests/src/main/java/plugin/PluginSupport.java @@ -5,16 +5,19 @@ /** * Shared helpers for the plugin conformance handlers (requirements 10-8..10-18). * - *

Every plugin captures the durable execution ARN from the invocation-start hook's info parameter and stamps it as - * a top-level {@code durableExecutionArn} field on every stdout JSON record, so the runner's execution-scoped - * CloudWatch filter ({@code $.durableExecutionArn = ""}) locates the records. These helpers only format that - * field and classify operation types reported by the real SDK; no behavior is fabricated here. + *

Every plugin captures the durable execution ARN from the invocation-start hook's info parameter and stamps it as a + * top-level {@code durableExecutionArn} field on every stdout JSON record, so the runner's execution-scoped CloudWatch + * filter ({@code $.durableExecutionArn = ""}) locates the records. These helpers only format that field and + * classify operation types reported by the real SDK; no behavior is fabricated here. */ final class PluginSupport { private PluginSupport() {} - /** Operation type token for step operations as reported by {@code OperationInfo#type()} (AWS SDK {@code OperationType}). */ + /** + * Operation type token for step operations as reported by {@code OperationInfo#type()} (AWS SDK + * {@code OperationType}). + */ static boolean isStep(String type) { return "STEP".equals(type); } diff --git a/conformance-tests/src/main/java/plugin/PluginWaitOperationHooks.java b/conformance-tests/src/main/java/plugin/PluginWaitOperationHooks.java index ab0fe671b..94b7ea5c3 100644 --- a/conformance-tests/src/main/java/plugin/PluginWaitOperationHooks.java +++ b/conformance-tests/src/main/java/plugin/PluginWaitOperationHooks.java @@ -16,8 +16,8 @@ * 10-10: Plugin operation-start and operation-end hooks fire for wait-type operations. * *

A single 2-second wait. The plugin, filtering to wait-type operations, logs operation-start when the wait's - * STARTED checkpoint is observed and operation-end with the terminal status. The type token is normalized to - * upper-case (WAIT). + * STARTED checkpoint is observed and operation-end with the terminal status. The type token is normalized to upper-case + * (WAIT). */ @SuppressWarnings("deprecation") public class PluginWaitOperationHooks extends DurableHandler { diff --git a/conformance-tests/template_map.yaml b/conformance-tests/template_map.yaml index 9819e4ba0..6228d216d 100644 --- a/conformance-tests/template_map.yaml +++ b/conformance-tests/template_map.yaml @@ -59,8 +59,6 @@ Resources: reason: "items-only form (no name): every Java context.map overload requires a name argument" - id: "9-6" reason: "throw-if-error rethrow: MapResult has no throw-if-error and map exposes no per-item futures" - - id: "9-13" - reason: "custom item namer: Java MapConfig has no item-namer field" - id: "9-14" reason: "custom per-item serdes: Java MapConfig has a single serDes (no separate item-level serdes distinct from the result serde)" - id: "9-19" @@ -95,6 +93,22 @@ Resources: RetentionPeriodInDays: 7 ExecutionTimeout: 300 + MapItemNamer: + Type: AWS::Serverless::Function + TestingMetadata: + TestDescription: ["9-13"] + Properties: + CodeUri: . + Handler: map.MapItemNamer + Description: Map with a custom item namer + Role: + Fn::GetAtt: + - DurableFunctionRole + - Arn + DurableConfig: + RetentionPeriodInDays: 7 + ExecutionTimeout: 300 + MapEmpty: Type: AWS::Serverless::Function TestingMetadata: diff --git a/sdk-integration-tests/src/test/java/software/amazon/lambda/durable/MapIntegrationTest.java b/sdk-integration-tests/src/test/java/software/amazon/lambda/durable/MapIntegrationTest.java index fef289c9b..45b129dbd 100644 --- a/sdk-integration-tests/src/test/java/software/amazon/lambda/durable/MapIntegrationTest.java +++ b/sdk-integration-tests/src/test/java/software/amazon/lambda/durable/MapIntegrationTest.java @@ -1888,4 +1888,303 @@ void testEmptyMapReplayUsesCheckpoint(NestingType nestingType, int events) { assertEquals(firstRunCount, executionCount.get(), "Map functions should not re-execute on replay"); assertEquals(events, result2.getHistoryEvents().size()); } + + @Test + void testMapWithItemNamerNamesChildOperations() { + var executionCount = new AtomicInteger(0); + var runner = LocalDurableTestRunner.create(String.class, (input, context) -> { + var result = context.map( + "named-map", + List.of("alpha", "beta"), + String.class, + (item, index, ctx) -> { + executionCount.incrementAndGet(); + return item.toUpperCase(); + }, + MapConfig.builder() + .maxConcurrency(1) + .itemNamer((item, index) -> "item-" + item) + .build()); + + assertTrue(result.allSucceeded()); + return String.join(",", result.results()); + }); + + var result = runner.runUntilComplete("test"); + assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus()); + assertEquals("ALPHA,BETA", result.getResult(String.class)); + + // The custom names must actually reach the child operations. + assertNotNull(result.getOperation("item-alpha"), "custom iteration name should be checkpointed"); + assertNotNull(result.getOperation("item-beta"), "custom iteration name should be checkpointed"); + + // ...and the default naming must no longer be used for those iterations. + assertNull(result.getOperation("named-map-iteration-0")); + assertNull(result.getOperation("named-map-iteration-1")); + + // Replay must resolve the custom-named operations without re-executing them. + var firstRunCount = executionCount.get(); + var replay = runner.run("test"); + assertEquals(ExecutionStatus.SUCCEEDED, replay.getStatus()); + assertEquals(firstRunCount, executionCount.get(), "Map functions should not re-execute on replay"); + assertNotNull(replay.getOperation("item-alpha")); + assertNotNull(replay.getOperation("item-beta")); + } + + @Test + void testMapWithoutItemNamerUsesDefaultIterationNames() { + var runner = LocalDurableTestRunner.create(String.class, (input, context) -> { + var result = context.map( + "default-named-map", + List.of("alpha", "beta"), + String.class, + (item, index, ctx) -> item.toUpperCase(), + MapConfig.builder().maxConcurrency(1).build()); + return String.join(",", result.results()); + }); + + var result = runner.runUntilComplete("test"); + assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus()); + + // Pins the documented default naming so the Javadoc and runtime cannot drift apart. + assertNotNull(result.getOperation("default-named-map-iteration-0")); + assertNotNull(result.getOperation("default-named-map-iteration-1")); + } + + @Test + void testItemNamerReturningDuplicateNamesStillProducesCorrectResults() { + var runner = LocalDurableTestRunner.create(String.class, (input, context) -> { + var result = context.map( + "dup-namer", + List.of("a", "b"), + String.class, + (item, index, ctx) -> item.toUpperCase(), + MapConfig.builder() + .maxConcurrency(1) + .itemNamer((item, index) -> "same") + .build()); + return String.join(",", result.results()); + }); + + var result = runner.runUntilComplete("test"); + assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus()); + + // Duplicate names are tolerated: operation ids are counter-derived, so results stay + // correctly ordered. The tradeoff is that name-based lookup becomes ambiguous. + assertEquals("A,B", result.getResult(String.class)); + assertEquals( + 2, + result.getOperations().stream() + .filter(op -> "same".equals(op.getName())) + .count(), + "both iterations should carry the duplicated name"); + assertNotNull(result.getOperation("same"), "lookup resolves to one of the duplicates"); + } + + @Test + void testItemNamerReturningNullFallsBackToDefaultNaming() { + var runner = LocalDurableTestRunner.create(String.class, (input, context) -> { + var result = context.map( + "null-namer", + List.of("a", "b"), + String.class, + (item, index, ctx) -> item.toUpperCase(), + MapConfig.builder() + .maxConcurrency(1) + .itemNamer((item, index) -> null) + .build()); + return String.join(",", result.results()); + }); + + var result = runner.runUntilComplete("test"); + assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus()); + assertEquals("A,B", result.getResult(String.class)); + + // A namer returning null must not checkpoint a nameless iteration; it falls back to the + // default naming so iterations stay addressable by name. + assertNotNull(result.getOperation("null-namer-iteration-0")); + assertNotNull(result.getOperation("null-namer-iteration-1")); + assertEquals( + 0, + result.getOperations().stream() + .filter(op -> op.getName() == null) + .count(), + "no iteration should be left without a name"); + } + + @Test + void testItemNamerReturningPartialNullFallsBackPerItem() { + var runner = LocalDurableTestRunner.create(String.class, (input, context) -> { + var result = context.map( + "partial-namer", + List.of("a", "b"), + String.class, + (item, index, ctx) -> item.toUpperCase(), + MapConfig.builder() + .maxConcurrency(1) + .itemNamer((item, index) -> index == 0 ? "named-first" : null) + .build()); + return String.join(",", result.results()); + }); + + var result = runner.runUntilComplete("test"); + assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus()); + + // The fallback is per-item, not all-or-nothing. + assertNotNull(result.getOperation("named-first")); + assertNotNull(result.getOperation("partial-namer-iteration-1")); + } + + @Test + void testItemNamerReturningInvalidNameFailsFast() { + // Empty, over-long and non-ASCII names are rejected the same way the map's own name is, + // instead of being sent to the backend and failing at checkpoint time. + assertInvalidItemNamerName(index -> ""); + assertInvalidItemNamerName(index -> "n\u00e9me-with-non-ascii"); + assertInvalidItemNamerName(index -> "x".repeat(1024)); + } + + @Test + void testTypedItemNamerReceivesDomainObjectWithoutCast() { + record Order(String id) {} + + var runner = LocalDurableTestRunner.create(String.class, (input, context) -> { + // MapConfig flows through map(...) and the namer sees Order, not Object. + MapConfig config = MapConfig.builder() + .maxConcurrency(1) + .itemNamer((order, index) -> "order-" + order.id()) + .build(); + + var result = context.map( + "typed-map", + List.of(new Order("A17"), new Order("B42")), + String.class, + (order, index, ctx) -> order.id().toLowerCase(), + config); + return String.join(",", result.results()); + }); + + var result = runner.runUntilComplete("test"); + assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus()); + assertEquals("a17,b42", result.getResult(String.class)); + assertNotNull(result.getOperation("order-A17")); + assertNotNull(result.getOperation("order-B42")); + } + + @Test + void testChangedItemNamerOnCachedReplayIsRejected() { + // A completed map with a small result replays from the payload: children are enqueued but + // never executed, so per-operation replay validation never compares their names. The map + // must still reject names that changed between invocations. + var invocation = new AtomicInteger(0); + var runner = LocalDurableTestRunner.create(String.class, (input, context) -> { + var suffix = invocation.get() == 0 ? "first" : "second"; + var result = context.map( + "changing-namer", + List.of("a", "b"), + String.class, + (item, index, ctx) -> item.toUpperCase(), + MapConfig.builder() + .maxConcurrency(1) + .itemNamer((item, index) -> item + "-" + suffix) + .build()); + return String.join(",", result.results()); + }); + + var first = runner.runUntilComplete("test"); + assertEquals(ExecutionStatus.SUCCEEDED, first.getStatus()); + assertNotNull(first.getOperation("a-first")); + assertNotNull(first.getOperation("b-first")); + + // Second invocation regenerates different names for the same checkpointed iterations. + invocation.incrementAndGet(); + var replay = runner.run("test"); + assertNotEquals( + ExecutionStatus.SUCCEEDED, + replay.getStatus(), + "a changed itemNamer must not replay silently under the checkpointed names"); + } + + @Test + void testStableItemNamerReplaysCleanlyFromCache() { + // Control for the test above: an unchanged namer must still replay from cache without error. + var runner = LocalDurableTestRunner.create(String.class, (input, context) -> { + var result = context.map( + "stable-namer", + List.of("a", "b"), + String.class, + (item, index, ctx) -> item.toUpperCase(), + MapConfig.builder() + .maxConcurrency(1) + .itemNamer((item, index) -> item + "-stable") + .build()); + return String.join(",", result.results()); + }); + + assertEquals(ExecutionStatus.SUCCEEDED, runner.runUntilComplete("test").getStatus()); + var replay = runner.run("test"); + assertEquals(ExecutionStatus.SUCCEEDED, replay.getStatus()); + assertNotNull(replay.getOperation("a-stable")); + assertNotNull(replay.getOperation("b-stable")); + } + + private void assertInvalidItemNamerName(java.util.function.Function nameFor) { + var runner = LocalDurableTestRunner.create(String.class, (input, context) -> { + var result = context.map( + "invalid-namer", + List.of("a"), + String.class, + (item, index, ctx) -> item.toUpperCase(), + MapConfig.builder() + .maxConcurrency(1) + .itemNamer((item, index) -> nameFor.apply(index)) + .build()); + return String.join(",", result.results()); + }); + + var result = runner.runUntilComplete("test"); + assertNotEquals( + ExecutionStatus.SUCCEEDED, result.getStatus(), "an invalid custom iteration name must not succeed"); + } + + @Test + void testCaughtItemNamerValidationLeavesNoHalfOpenMap() { + var runner = LocalDurableTestRunner.create(String.class, (input, context) -> { + // The handler swallows the validation failure, which is the case that would previously + // strand a STARTED map with an unbalanced plugin lifecycle. + assertThrows( + IllegalArgumentException.class, + () -> context.map( + "caught-invalid", + List.of("a", "b"), + String.class, + (item, index, ctx) -> item.toUpperCase(), + MapConfig.builder() + .maxConcurrency(1) + .itemNamer((item, index) -> "") + .build())); + + // The execution must still be able to make progress afterwards. + var recovered = context.map( + "recovered", + List.of("a", "b"), + String.class, + (item, index, ctx) -> item.toUpperCase(), + MapConfig.builder().maxConcurrency(1).build()); + return String.join(",", recovered.results()); + }); + + var result = runner.runUntilComplete("test"); + assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus()); + assertEquals("A,B", result.getResult(String.class)); + + // The rejected map must not have checkpointed anything at all. + assertNull(result.getOperation("caught-invalid"), "failed map should not be checkpointed"); + assertNotNull(result.getOperation("recovered")); + + // Replay must also succeed, proving no STARTED operation was left behind. + var replay = runner.run("test"); + assertEquals(ExecutionStatus.SUCCEEDED, replay.getStatus()); + assertNull(replay.getOperation("caught-invalid")); + } } diff --git a/sdk/src/main/java/software/amazon/lambda/durable/DurableContext.java b/sdk/src/main/java/software/amazon/lambda/durable/DurableContext.java index ce6eb7070..4a159f4b1 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/DurableContext.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/DurableContext.java @@ -482,24 +482,37 @@ default MapResult map(String name, Collection items, Class resul items, TypeToken.get(resultType), function, - MapConfig.builder().build()) + MapConfig.builder().build()) .get(); } default MapResult map( - String name, Collection items, Class resultType, MapFunction function, MapConfig config) { + String name, + Collection items, + Class resultType, + MapFunction function, + MapConfig config) { return mapAsync(name, items, TypeToken.get(resultType), function, config) .get(); } default MapResult map( String name, Collection items, TypeToken resultType, MapFunction function) { - return mapAsync(name, items, resultType, function, MapConfig.builder().build()) + return mapAsync( + name, + items, + resultType, + function, + MapConfig.builder().build()) .get(); } default MapResult map( - String name, Collection items, TypeToken resultType, MapFunction function, MapConfig config) { + String name, + Collection items, + TypeToken resultType, + MapFunction function, + MapConfig config) { return mapAsync(name, items, resultType, function, config).get(); } @@ -510,21 +523,30 @@ default DurableFuture> mapAsync( items, TypeToken.get(resultType), function, - MapConfig.builder().build()); + MapConfig.builder().build()); } default DurableFuture> mapAsync( - String name, Collection items, Class resultType, MapFunction function, MapConfig config) { + String name, + Collection items, + Class resultType, + MapFunction function, + MapConfig config) { return mapAsync(name, items, TypeToken.get(resultType), function, config); } default DurableFuture> mapAsync( String name, Collection items, TypeToken resultType, MapFunction function) { - return mapAsync(name, items, resultType, function, MapConfig.builder().build()); + return mapAsync( + name, items, resultType, function, MapConfig.builder().build()); } DurableFuture> mapAsync( - String name, Collection items, TypeToken resultType, MapFunction function, MapConfig config); + String name, + Collection items, + TypeToken resultType, + MapFunction function, + MapConfig config); /** * Creates a {@link ParallelDurableFuture} for executing multiple branches concurrently with default config diff --git a/sdk/src/main/java/software/amazon/lambda/durable/config/MapConfig.java b/sdk/src/main/java/software/amazon/lambda/durable/config/MapConfig.java index d92572617..212c4c891 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/config/MapConfig.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/config/MapConfig.java @@ -3,24 +3,29 @@ package software.amazon.lambda.durable.config; import java.util.Objects; +import java.util.function.BiFunction; import software.amazon.lambda.durable.serde.SerDes; /** * Configuration for map operations. * *

Defaults to lenient completion (all items run regardless of failures) and unlimited concurrency. + * + * @param the type of the map input items */ -public class MapConfig { +public class MapConfig { private final Integer maxConcurrency; private final CompletionConfig completionConfig; private final SerDes serDes; private final NestingType nestingType; + private final BiFunction itemNamer; - private MapConfig(Builder builder) { + private MapConfig(Builder builder) { this.maxConcurrency = Objects.requireNonNullElse(builder.maxConcurrency, Integer.MAX_VALUE); this.completionConfig = Objects.requireNonNullElse(builder.completionConfig, CompletionConfig.allCompleted()); this.nestingType = Objects.requireNonNullElse(builder.nestingType, NestingType.NESTED); this.serDes = builder.serDes; + this.itemNamer = builder.itemNamer; } /** @return max concurrent items, or null for unlimited */ @@ -43,28 +48,65 @@ public NestingType nestingType() { return nestingType; } - public static Builder builder() { - return new Builder(); + /** + * Returns the item namer function, which generates custom names for each map iteration. + * + *

When provided, the namer is called for each item with the item value and its index. The returned string is + * used as the operation name for that iteration, replacing the default {@code "-iteration-N"} naming. + * + *

The returned name must satisfy the same constraints as any other operation name: non-empty, within the maximum + * operation-name length, and printable ASCII only. An invalid name fails fast with {@link IllegalArgumentException} + * before the map checkpoints anything. Returning {@code null} is permitted and falls back to the default naming for + * that iteration. + * + *

The namer must be deterministic: replay regenerates iteration names and compares them against the checkpointed + * names, so a namer whose output varies between invocations risks a non-deterministic replay failure. + * + * @return the item namer function, or null if not set + */ + public BiFunction itemNamer() { + return itemNamer; + } + + /** + * Creates a new builder for {@code MapConfig}. All fields are optional. + * + * @param the type of the map input items + * @return a new builder instance + */ + public static Builder builder() { + return new Builder<>(); } - public Builder toBuilder() { - return new Builder() + /** + * Returns a new builder initialized with the values from this config. + * + * @return a new builder pre-populated with this config's values + */ + public Builder toBuilder() { + return MapConfig.builder() .maxConcurrency(maxConcurrency) .completionConfig(completionConfig) .serDes(serDes) - .nestingType(nestingType); + .nestingType(nestingType) + .itemNamer(itemNamer); } - /** Builder for creating MapConfig instances. */ - public static class Builder { + /** + * Builder for creating MapConfig instances. + * + * @param the type of the map input items + */ + public static class Builder { public NestingType nestingType; private Integer maxConcurrency; private CompletionConfig completionConfig; private SerDes serDes; + private BiFunction itemNamer; private Builder() {} - public Builder maxConcurrency(Integer maxConcurrency) { + public Builder maxConcurrency(Integer maxConcurrency) { if (maxConcurrency != null && maxConcurrency < 1) { throw new IllegalArgumentException("maxConcurrency must be at least 1, got: " + maxConcurrency); } @@ -78,7 +120,7 @@ public Builder maxConcurrency(Integer maxConcurrency) { * @param completionConfig the completion configuration (default: {@link CompletionConfig#allCompleted()}) * @return this builder for method chaining */ - public Builder completionConfig(CompletionConfig completionConfig) { + public Builder completionConfig(CompletionConfig completionConfig) { this.completionConfig = completionConfig; return this; } @@ -89,7 +131,7 @@ public Builder completionConfig(CompletionConfig completionConfig) { * @param serDes the serializer to use * @return this builder for method chaining */ - public Builder serDes(SerDes serDes) { + public Builder serDes(SerDes serDes) { this.serDes = serDes; return this; } @@ -100,13 +142,32 @@ public Builder serDes(SerDes serDes) { * @param nestingType the nesting type (default: {@link NestingType#NESTED}) * @return this builder for method chaining */ - public Builder nestingType(NestingType nestingType) { + public Builder nestingType(NestingType nestingType) { this.nestingType = nestingType; return this; } - public MapConfig build() { - return new MapConfig(this); + /** + * Sets the item namer function for generating custom iteration names. + * + *

The namer receives the item and its index, and returns a string to use as the operation name for that + * iteration. If the namer is null, or returns null for an item, the default {@code "-iteration-N"} + * naming is used for that iteration. + * + *

A non-null name must be non-empty, within the maximum operation-name length, and printable ASCII only; + * otherwise an {@link IllegalArgumentException} is thrown before the map checkpoints anything. The namer must + * also be deterministic across replays. + * + * @param itemNamer the item namer function, or null to use default naming + * @return this builder for method chaining + */ + public Builder itemNamer(BiFunction itemNamer) { + this.itemNamer = itemNamer; + return this; + } + + public MapConfig build() { + return new MapConfig<>(this); } } } diff --git a/sdk/src/main/java/software/amazon/lambda/durable/context/DurableContextImpl.java b/sdk/src/main/java/software/amazon/lambda/durable/context/DurableContextImpl.java index d75e5b8bd..5f3b73dfe 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/context/DurableContextImpl.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/context/DurableContextImpl.java @@ -254,7 +254,11 @@ private DurableFuture runInChildContextAsync( @Override public DurableFuture> mapAsync( - String name, Collection items, TypeToken resultType, MapFunction function, MapConfig config) { + String name, + Collection items, + TypeToken resultType, + MapFunction function, + MapConfig config) { Objects.requireNonNull(items, "items cannot be null"); Objects.requireNonNull(function, "function cannot be null"); Objects.requireNonNull(resultType, "resultType cannot be null"); diff --git a/sdk/src/main/java/software/amazon/lambda/durable/execution/ExecutionManager.java b/sdk/src/main/java/software/amazon/lambda/durable/execution/ExecutionManager.java index 1c45cb0d6..b7225cd41 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/execution/ExecutionManager.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/execution/ExecutionManager.java @@ -205,6 +205,19 @@ public Operation getExecutionOperation() { return executionOp; } + /** + * Looks up a checkpointed operation without touching replay state. + * + *

Unlike {@link #getOperationAndUpdateReplayState(String)} this performs no REPLAY to EXECUTION transition, so + * it is safe to call for read-only validation of operations that are not being executed. + * + * @param operationId the globally unique operation ID + * @return the checkpointed operation, or null if not present + */ + public Operation peekOperation(String operationId) { + return operationStorage.get(operationId); + } + /** * Checks whether there are any cached operations for the given parent context ID. Used to initialize per-context * replay state — a context starts in replay mode if the ExecutionManager has cached operations belonging to it. diff --git a/sdk/src/main/java/software/amazon/lambda/durable/operation/MapOperation.java b/sdk/src/main/java/software/amazon/lambda/durable/operation/MapOperation.java index 8caf2a647..371bcb4b9 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/operation/MapOperation.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/operation/MapOperation.java @@ -6,6 +6,8 @@ import java.util.ArrayList; import java.util.Collections; import java.util.List; +import java.util.Objects; +import java.util.function.BiFunction; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import software.amazon.awssdk.services.lambda.model.ContextOptions; @@ -17,6 +19,7 @@ import software.amazon.lambda.durable.config.CompletionConfig; import software.amazon.lambda.durable.config.MapConfig; import software.amazon.lambda.durable.context.DurableContextImpl; +import software.amazon.lambda.durable.exception.NonDeterministicExecutionException; import software.amazon.lambda.durable.exception.UnrecoverableDurableExecutionException; import software.amazon.lambda.durable.execution.SuspendExecutionException; import software.amazon.lambda.durable.model.ConcurrencyCompletionStatus; @@ -25,6 +28,7 @@ import software.amazon.lambda.durable.model.OperationSubType; import software.amazon.lambda.durable.serde.SerDes; import software.amazon.lambda.durable.util.ExceptionHelper; +import software.amazon.lambda.durable.util.ParameterValidator; /** * Executes a map operation: applies a function to each item in a collection concurrently, with each item running in its @@ -46,6 +50,8 @@ public class MapOperation extends ConcurrencyOperation> { private final DurableContext.MapFunction function; private final TypeToken itemResultType; private final SerDes serDes; + private final BiFunction itemNamer; + private final List iterationNames; private volatile MapResult cachedResult; public MapOperation( @@ -53,7 +59,7 @@ public MapOperation( List items, DurableContext.MapFunction function, TypeToken itemResultType, - MapConfig config, + MapConfig config, DurableContextImpl durableContext) { super( operationIdentifier, @@ -73,17 +79,66 @@ public MapOperation( this.function = function; this.itemResultType = itemResultType; this.serDes = config.serDes(); + this.itemNamer = config.itemNamer(); + // Resolved here, before execute() checkpoints START and fires the start plugin hook. If the + // namer throws or yields an invalid name, failing in the constructor leaves no half-open + // operation behind; failing later would leave a permanently STARTED map and an unbalanced + // plugin lifecycle whenever the caller catches the exception. + this.iterationNames = resolveIterationNames(); + } + + /** + * Resolves the operation name for every iteration. + * + *

A configured itemNamer takes precedence, but a null result falls back to the default naming rather than + * checkpointing a nameless iteration. Custom names skip the validation that {@code DurableContextImpl.mapAsync} + * applies to the map's own name, so they are validated here. + */ + private List resolveIterationNames() { + var names = new ArrayList(items.size()); + var branchPrefix = getName() == null ? "map-iteration-" : getName() + "-iteration-"; + for (int i = 0; i < items.size(); i++) { + var name = itemNamer == null ? null : itemNamer.apply(items.get(i), i); + if (name == null) { + name = branchPrefix + i; + } else { + ParameterValidator.validateOperationName(name); + } + names.add(name); + } + return List.copyOf(names); } private void addAllItems() { addUnskippedItems(Collections.nCopies(items.size(), null)); } + /** + * Compares the freshly generated iteration names against the checkpointed child operations. + * + *

When a completed map's result is small enough to be replayed from the payload, its children are enqueued but + * never executed, so the per-operation {@code validateReplay} name check never runs. Without this, a namer whose + * output changed between invocations would replay silently under the old names. + */ + private void validateIterationNamesAgainstCheckpoint() { + for (var branch : getBranches()) { + var checkpointed = executionManager.peekOperation(branch.getOperationId()); + if (checkpointed == null || checkpointed.name() == null) { + continue; + } + if (!Objects.equals(checkpointed.name(), branch.getName())) { + throw terminateExecution(new NonDeterministicExecutionException(String.format( + "Map iteration name mismatch for \"%s\". Expected \"%s\", got \"%s\". " + + "The map's itemNamer must be deterministic across replays.", + branch.getOperationId(), checkpointed.name(), branch.getName()))); + } + } + } + private void addUnskippedItems(List resultItems) { // Enqueue all items first. // If the map is completed when replaying, mapResult != null and the items that have been skipped // will be skipped during replay. - var branchPrefix = getName() == null ? "map-iteration-" : getName() + "-iteration-"; for (int i = 0; i < items.size(); i++) { var index = i; var item = items.get(i); @@ -92,7 +147,7 @@ private void addUnskippedItems(List resultItems) var skip = status == MapResult.MapResultItem.Status.SKIPPED; enqueueItem( - branchPrefix + i, + iterationNames.get(i), childCtx -> function.apply(item, index, childCtx), itemResultType, serDes, @@ -150,6 +205,7 @@ protected void replay(Operation existing) { throw terminateExecutionWithIllegalDurableOperationException( "Missing result in completed Map operation"); } + validateIterationNamesAgainstCheckpoint(); if (Boolean.TRUE.equals(existing.contextDetails().replayChildren())) { // Large result: re-execute children to reconstruct MapResult var expected = new ExpectedCompletionStatus( diff --git a/sdk/src/test/java/software/amazon/lambda/durable/config/MapConfigItemNamerTest.java b/sdk/src/test/java/software/amazon/lambda/durable/config/MapConfigItemNamerTest.java new file mode 100644 index 000000000..6aa3e15a2 --- /dev/null +++ b/sdk/src/test/java/software/amazon/lambda/durable/config/MapConfigItemNamerTest.java @@ -0,0 +1,132 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.config; + +import static org.junit.jupiter.api.Assertions.*; + +import java.util.function.BiFunction; +import org.junit.jupiter.api.Test; + +/** Tests for {@link MapConfig} itemNamer functionality. */ +class MapConfigItemNamerTest { + + @Test + void builder_withoutItemNamer_returnsNullItemNamer() { + MapConfig config = MapConfig.builder().build(); + assertNull(config.itemNamer()); + } + + @Test + void builder_withItemNamer_returnsConfiguredItemNamer() { + BiFunction namer = (item, idx) -> "item-" + idx; + MapConfig config = MapConfig.builder().itemNamer(namer).build(); + assertSame(namer, config.itemNamer()); + } + + @Test + void itemNamer_calledWithItemAndIndex() { + BiFunction namer = (item, idx) -> "name-" + item + "-" + idx; + MapConfig config = MapConfig.builder().itemNamer(namer).build(); + + String result = config.itemNamer().apply("test-item", 42); + assertEquals("name-test-item-42", result); + } + + @Test + void toBuilder_preservesItemNamer() { + BiFunction namer = (item, idx) -> "custom-" + idx; + MapConfig original = + MapConfig.builder().maxConcurrency(5).itemNamer(namer).build(); + + MapConfig rebuilt = original.toBuilder().build(); + assertEquals(5, rebuilt.maxConcurrency()); + assertSame(namer, rebuilt.itemNamer()); + } + + @Test + void toBuilder_canUpdateItemNamer() { + BiFunction namer1 = (item, idx) -> "first-" + idx; + BiFunction namer2 = (item, idx) -> "second-" + idx; + + MapConfig config = MapConfig.builder().itemNamer(namer1).build(); + + MapConfig updated = config.toBuilder().itemNamer(namer2).build(); + + assertSame(namer2, updated.itemNamer()); + } + + @Test + void toBuilder_canRemoveItemNamer() { + BiFunction namer = (item, idx) -> "name-" + idx; + MapConfig config = MapConfig.builder().itemNamer(namer).build(); + + MapConfig withoutNamer = config.toBuilder().itemNamer(null).build(); + + assertNull(withoutNamer.itemNamer()); + } + + @Test + void builder_withAllFields_includesItemNamer() { + MapConfig config = MapConfig.builder() + .maxConcurrency(3) + .nestingType(NestingType.FLAT) + .itemNamer((item, idx) -> "iter-" + idx) + .build(); + + assertEquals(3, config.maxConcurrency()); + assertEquals(NestingType.FLAT, config.nestingType()); + assertNotNull(config.itemNamer()); + assertEquals("iter-7", config.itemNamer().apply("anything", 7)); + } + + @Test + void itemNamer_withDomainType_needsNoCast() { + record Order(String id, int quantity) {} + + // The point of parameterizing MapConfig: the namer receives the domain type directly, + // so no cast from Object is needed. + MapConfig config = MapConfig.builder() + .itemNamer((order, idx) -> "order-" + order.id()) + .build(); + + assertEquals("order-A17", config.itemNamer().apply(new Order("A17", 3), 0)); + } + + @Test + void itemNamer_withDifferentItemTypes() { + // String items + BiFunction stringNamer = (item, idx) -> "str-" + item; + MapConfig stringConfig = + MapConfig.builder().itemNamer(stringNamer).build(); + assertEquals("str-hello", stringConfig.itemNamer().apply("hello", 0)); + + // Integer items + BiFunction intNamer = (item, idx) -> "num-" + item; + MapConfig intConfig = MapConfig.builder().itemNamer(intNamer).build(); + assertEquals("num-123", intConfig.itemNamer().apply(123, 1)); + + // Custom object items + record User(String id, String name) {} + BiFunction userNamer = (item, idx) -> { + User user = (User) item; + return "user-" + user.id(); + }; + MapConfig userConfig = MapConfig.builder().itemNamer(userNamer).build(); + assertEquals("user-u123", userConfig.itemNamer().apply(new User("u123", "Alice"), 2)); + } + + @Test + void itemNamer_inheritsOtherConfigFields() { + MapConfig base = MapConfig.builder() + .maxConcurrency(10) + .nestingType(NestingType.FLAT) + .build(); + + BiFunction namer = (item, idx) -> "named-" + idx; + MapConfig withNamer = base.toBuilder().itemNamer(namer).build(); + + assertEquals(10, withNamer.maxConcurrency()); + assertEquals(NestingType.FLAT, withNamer.nestingType()); + assertSame(namer, withNamer.itemNamer()); + } +} diff --git a/sdk/src/test/java/software/amazon/lambda/durable/operation/MapOperationItemNamerTest.java b/sdk/src/test/java/software/amazon/lambda/durable/operation/MapOperationItemNamerTest.java new file mode 100644 index 000000000..428d9be2b --- /dev/null +++ b/sdk/src/test/java/software/amazon/lambda/durable/operation/MapOperationItemNamerTest.java @@ -0,0 +1,99 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.operation; + +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.util.List; +import java.util.function.BiFunction; +import org.junit.jupiter.api.Test; +import software.amazon.lambda.durable.TypeToken; +import software.amazon.lambda.durable.config.MapConfig; +import software.amazon.lambda.durable.context.DurableContextImpl; +import software.amazon.lambda.durable.execution.ExecutionManager; +import software.amazon.lambda.durable.model.OperationIdentifier; +import software.amazon.lambda.durable.model.OperationSubType; + +/** Integration tests for MapOperation with itemNamer functionality. */ +class MapOperationItemNamerTest { + + @Test + void mapOperation_withItemNamer_constructsSuccessfully() { + // Setup minimal mock context with execution manager + DurableContextImpl mockContext = mock(DurableContextImpl.class); + ExecutionManager mockExecutionManager = mock(ExecutionManager.class); + when(mockContext.getExecutionManager()).thenReturn(mockExecutionManager); + + // Create custom item namer + BiFunction itemNamer = (item, idx) -> "process-" + item + "-" + idx; + + MapConfig config = + MapConfig.builder().itemNamer(itemNamer).maxConcurrency(2).build(); + + List items = List.of("order-101", "order-102", "order-103"); + + // Create MapOperation - should construct without errors + MapOperation operation = new MapOperation<>( + OperationIdentifier.of("map-1", "process_orders", OperationSubType.MAP), + items, + (item, index, ctx) -> "processed-" + item, + TypeToken.get(String.class), + config, + mockContext); + + // Verify the operation was created + assertNotNull(operation); + // Config should have the item namer + assertSame(itemNamer, config.itemNamer()); + } + + @Test + void mapOperation_withoutItemNamer_constructsSuccessfully() { + // Default config (no itemNamer) + MapConfig config = MapConfig.builder().build(); + + // Should not throw NPE + assertNull(config.itemNamer()); + + // Operation should be constructible + DurableContextImpl mockContext = mock(DurableContextImpl.class); + ExecutionManager mockExecutionManager = mock(ExecutionManager.class); + when(mockContext.getExecutionManager()).thenReturn(mockExecutionManager); + + MapOperation operation = new MapOperation<>( + OperationIdentifier.of("map-1", "test", OperationSubType.MAP), + List.of("a", "b"), + (item, index, ctx) -> item, + TypeToken.get(String.class), + config, + mockContext); + + assertNotNull(operation); + } + + @Test + void mapOperation_withNullItemNamer_constructsSuccessfully() { + // Explicitly null itemNamer + MapConfig config = MapConfig.builder().itemNamer(null).build(); + + assertNull(config.itemNamer()); + + DurableContextImpl mockContext = mock(DurableContextImpl.class); + ExecutionManager mockExecutionManager = mock(ExecutionManager.class); + when(mockContext.getExecutionManager()).thenReturn(mockExecutionManager); + + MapOperation operation = new MapOperation<>( + OperationIdentifier.of("map-1", "test", OperationSubType.MAP), + List.of("x", "y"), + (item, index, ctx) -> item, + TypeToken.get(String.class), + config, + mockContext); + + assertNotNull(operation); + } +}