From 4237a80fe36b1d0c90efcf8026cd18566d4300e0 Mon Sep 17 00:00:00 2001 From: nvasiu Date: Fri, 28 Aug 2026 23:06:50 +0000 Subject: [PATCH] feat(sdk): add distributed map operation --- .../durable/DistributedMapHandlers.java | 265 +++++++++++++++ .../amazon/lambda/durable/DurableContext.java | 141 ++++++++ .../lambda/durable/config/CsvDelimiter.java | 23 ++ .../lambda/durable/config/CsvFormat.java | 45 +++ .../DistributedMapCompletionConfig.java | 45 +++ .../durable/config/DistributedMapConfig.java | 91 +++++ .../config/DistributedMapDestination.java | 44 +++ .../DistributedMapDestinationConfig.java | 22 ++ .../config/DistributedMapProcessor.java | 123 +++++++ .../durable/config/DistributedMapSource.java | 212 ++++++++++++ .../durable/config/FailureDestination.java | 17 + .../durable/config/ProcessorRetryConfig.java | 76 +++++ .../durable/config/SuccessDestination.java | 17 + .../durable/context/DurableContextImpl.java | 87 +++++ .../exception/DistributedMapError.java | 93 +++++ .../exception/DistributedMapException.java | 18 + .../model/DistributedMapCompletionReason.java | 47 +++ .../model/DistributedMapItemError.java | 12 + .../durable/model/DistributedMapResult.java | 112 ++++++ .../model/DistributedMapResultItem.java | 23 ++ .../durable/model/DistributedMapStatus.java | 37 ++ .../durable/model/DistributedMapSummary.java | 36 ++ .../durable/model/OperationSubType.java | 4 +- .../lambda/durable/model/ReaderPage.java | 15 + .../operation/DistributedMapOperation.java | 79 +++++ .../durable/operation/DistributedMapWire.java | 318 ++++++++++++++++++ .../util/DistributedMapValidation.java | 62 ++++ 27 files changed, 2063 insertions(+), 1 deletion(-) create mode 100644 sdk/src/main/java/software/amazon/lambda/durable/DistributedMapHandlers.java create mode 100644 sdk/src/main/java/software/amazon/lambda/durable/config/CsvDelimiter.java create mode 100644 sdk/src/main/java/software/amazon/lambda/durable/config/CsvFormat.java create mode 100644 sdk/src/main/java/software/amazon/lambda/durable/config/DistributedMapCompletionConfig.java create mode 100644 sdk/src/main/java/software/amazon/lambda/durable/config/DistributedMapConfig.java create mode 100644 sdk/src/main/java/software/amazon/lambda/durable/config/DistributedMapDestination.java create mode 100644 sdk/src/main/java/software/amazon/lambda/durable/config/DistributedMapDestinationConfig.java create mode 100644 sdk/src/main/java/software/amazon/lambda/durable/config/DistributedMapProcessor.java create mode 100644 sdk/src/main/java/software/amazon/lambda/durable/config/DistributedMapSource.java create mode 100644 sdk/src/main/java/software/amazon/lambda/durable/config/FailureDestination.java create mode 100644 sdk/src/main/java/software/amazon/lambda/durable/config/ProcessorRetryConfig.java create mode 100644 sdk/src/main/java/software/amazon/lambda/durable/config/SuccessDestination.java create mode 100644 sdk/src/main/java/software/amazon/lambda/durable/exception/DistributedMapError.java create mode 100644 sdk/src/main/java/software/amazon/lambda/durable/exception/DistributedMapException.java create mode 100644 sdk/src/main/java/software/amazon/lambda/durable/model/DistributedMapCompletionReason.java create mode 100644 sdk/src/main/java/software/amazon/lambda/durable/model/DistributedMapItemError.java create mode 100644 sdk/src/main/java/software/amazon/lambda/durable/model/DistributedMapResult.java create mode 100644 sdk/src/main/java/software/amazon/lambda/durable/model/DistributedMapResultItem.java create mode 100644 sdk/src/main/java/software/amazon/lambda/durable/model/DistributedMapStatus.java create mode 100644 sdk/src/main/java/software/amazon/lambda/durable/model/DistributedMapSummary.java create mode 100644 sdk/src/main/java/software/amazon/lambda/durable/model/ReaderPage.java create mode 100644 sdk/src/main/java/software/amazon/lambda/durable/operation/DistributedMapOperation.java create mode 100644 sdk/src/main/java/software/amazon/lambda/durable/operation/DistributedMapWire.java create mode 100644 sdk/src/main/java/software/amazon/lambda/durable/util/DistributedMapValidation.java diff --git a/sdk/src/main/java/software/amazon/lambda/durable/DistributedMapHandlers.java b/sdk/src/main/java/software/amazon/lambda/durable/DistributedMapHandlers.java new file mode 100644 index 000000000..28f8ccca4 --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/DistributedMapHandlers.java @@ -0,0 +1,265 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable; + +import com.amazonaws.services.lambda.runtime.RequestHandler; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.function.BiConsumer; +import java.util.function.BiFunction; +import java.util.function.Consumer; +import java.util.function.Function; +import software.amazon.lambda.durable.config.CompletionConfig; +import software.amazon.lambda.durable.config.DistributedMapProcessor.ResponseMode; +import software.amazon.lambda.durable.config.MapConfig; +import software.amazon.lambda.durable.execution.DurableExecutor; +import software.amazon.lambda.durable.model.DurableExecutionInput; +import software.amazon.lambda.durable.model.DurableExecutionOutput; +import software.amazon.lambda.durable.model.MapResult; +import software.amazon.lambda.durable.model.ReaderPage; +import software.amazon.lambda.durable.serde.JacksonSerDes; +import software.amazon.lambda.durable.serde.SerDes; + +/** Authoring helpers that wrap a customer function into a distributed map processor or reader Lambda handler. */ +public final class DistributedMapHandlers { + private static final SerDes JSON = new JacksonSerDes(); + private static final TypeToken OBJECT_TYPE = TypeToken.get(Object.class); + private static final TypeToken> EVENT_TYPE = new TypeToken>() {}; + private static final int READER_STATE_LIMIT = 32 * 1024; + private static final String ITEMS_OP_NAME = "distributed-map-items"; + + private DistributedMapHandlers() {} + + private static void validateItemReport(ResponseMode report) { + if (report != ResponseMode.REPORT_ITEM_RESULTS && report != ResponseMode.REPORT_FAILED_ITEMS) { + throw new IllegalArgumentException( + "item handler report mode must be REPORT_ITEM_RESULTS or REPORT_FAILED_ITEMS, got: " + report); + } + } + + /** Wraps a per-item function as a processor handler. Use REPORT_ITEM_RESULTS or REPORT_FAILED_ITEMS. */ + public static RequestHandler, Map> createDistributedMapItemHandler( + Function func, + TypeToken itemType, + SerDes itemSerDes, + SerDes resultSerDes, + ResponseMode report, + int concurrency) { + validateItemReport(report); + var inSerdes = itemSerDes != null ? itemSerDes : JSON; + var outSerdes = resultSerDes != null ? resultSerDes : JSON; + return (event, context) -> { + var records = records(event); + var workers = concurrency > 0 ? concurrency : Math.max(1, records.size()); + var outputs = new Object[records.size()]; + var errors = new Throwable[records.size()]; + var pool = Executors.newFixedThreadPool(workers); + try { + var futures = new ArrayList>(); + for (var i = 0; i < records.size(); i++) { + var index = i; + futures.add(pool.submit(() -> { + I item = toItem(inSerdes, records.get(index).get("body"), itemType); + outputs[index] = func.apply(item); + return null; + })); + } + for (var i = 0; i < futures.size(); i++) { + try { + futures.get(i).get(); + } catch (ExecutionException e) { + errors[i] = e.getCause() != null ? e.getCause() : e; + } + } + } catch (InterruptedException e) { + pool.shutdownNow(); + Thread.currentThread().interrupt(); + throw new IllegalStateException("Distributed map item handler interrupted", e); + } finally { + pool.shutdown(); + } + + var results = new ArrayList>(); + var failures = new ArrayList>(); + for (var i = 0; i < records.size(); i++) { + var itemId = (String) records.get(i).get("itemId"); + if (errors[i] != null) { + failures.add(errorEntry(itemId, errors[i].getClass().getName(), errors[i].getMessage())); + } else if (report == ResponseMode.REPORT_ITEM_RESULTS) { + results.add(resultEntry(itemId, toJsonValue(outSerdes, outputs[i]))); + } + } + return itemResponse(report, results, failures); + }; + } + + /** Wraps a whole-batch function as a processor handler. Returning succeeds every item, throwing fails the batch. */ + public static RequestHandler, Object> createDistributedMapBatchHandler( + Consumer> func, TypeToken itemType, SerDes itemSerDes) { + var serdes = itemSerDes != null ? itemSerDes : JSON; + return (event, context) -> { + func.accept(toItems(serdes, event, itemType)); + return null; + }; + } + + /** Wraps a reader function as a source handler. A null nextState signals the source is exhausted. */ + public static RequestHandler, Map> createDistributedMapReader( + Function> func, TypeToken stateType, SerDes stateSerDes) { + var serdes = stateSerDes != null ? stateSerDes : JSON; + return (event, context) -> { + if (!(event.get("maxItems") instanceof Number maxItemsValue)) { + throw new IllegalStateException("expected a distributed map reader event carrying a numeric 'maxItems' (got keys: " + + event.keySet() + "), the function must be registered as a distributed map reader"); + } + var rawState = (String) event.get("state"); + S state = rawState != null ? serdes.deserialize(rawState, stateType) : null; + var maxItems = maxItemsValue.intValue(); + + var page = func.apply(state); + if (page.items().size() > maxItems) { + throw new IllegalStateException( + "reader returned " + page.items().size() + " items, exceeding maxItems " + maxItems); + } + var response = new LinkedHashMap(); + response.put("items", page.items()); + if (page.nextState() != null) { + var nextState = serdes.serialize(page.nextState()); + if (nextState.getBytes(StandardCharsets.UTF_8).length > READER_STATE_LIMIT) { + throw new IllegalStateException( + "reader nextState exceeds the " + (READER_STATE_LIMIT / 1024) + " KB limit"); + } + response.put("nextState", nextState); + } + return response; + }; + } + + /** Durable variant of the item handler. The function receives the per-item DurableContext. */ + public static + RequestHandler + createDistributedMapItemHandlerWithDurableExecution( + BiFunction func, + TypeToken itemType, + SerDes itemSerDes, + SerDes resultSerDes, + ResponseMode report) { + validateItemReport(report); + var inSerdes = itemSerDes != null ? itemSerDes : JSON; + var outSerdes = resultSerDes != null ? resultSerDes : JSON; + return DurableExecutor.wrap( + EVENT_TYPE, + (event, ctx) -> { + var records = records(event); + var bodies = new ArrayList(records.size()); + for (var record : records) { + bodies.add(record.get("body")); + } + MapResult batch = ctx.map( + ITEMS_OP_NAME, + bodies, + OBJECT_TYPE, + (body, index, mapContext) -> + (Object) func.apply(mapContext, toItem(inSerdes, body, itemType)), + MapConfig.builder() + .completionConfig(CompletionConfig.allCompleted()) + .build()); + + var results = new ArrayList>(); + var failures = new ArrayList>(); + for (var i = 0; i < records.size(); i++) { + var itemId = (String) records.get(i).get("itemId"); + var item = batch.getItem(i); + if (item.status() == MapResult.MapResultItem.Status.SUCCEEDED) { + if (report == ResponseMode.REPORT_ITEM_RESULTS) { + results.add(resultEntry(itemId, toJsonValue(outSerdes, item.result()))); + } + } else { + var error = item.error(); + failures.add(errorEntry( + itemId, + error != null ? error.errorType() : "", + error != null ? error.errorMessage() : "")); + } + } + return itemResponse(report, results, failures); + }, + DurableConfig.defaultConfig()); + } + + /** Durable variant of the batch handler. The function receives the DurableContext. */ + public static + RequestHandler + createDistributedMapBatchHandlerWithDurableExecution( + BiConsumer> func, TypeToken itemType, SerDes itemSerDes) { + var serdes = itemSerDes != null ? itemSerDes : JSON; + return DurableExecutor.wrap( + EVENT_TYPE, + (event, ctx) -> { + func.accept(ctx, toItems(serdes, event, itemType)); + return null; + }, + DurableConfig.defaultConfig()); + } + + @SuppressWarnings("unchecked") + private static List> records(Map event) { + if (!(event.get("records") instanceof List)) { + throw new IllegalStateException("expected a distributed map processor event carrying a 'records' array (got keys: " + + event.keySet() + "), the function must be triggered by a distributed map, not invoked directly"); + } + return (List>) event.get("records"); + } + + private static List toItems(SerDes serdes, Map event, TypeToken itemType) { + var records = records(event); + var items = new ArrayList(records.size()); + for (var record : records) { + items.add(toItem(serdes, record.get("body"), itemType)); + } + return items; + } + + private static I toItem(SerDes serdes, Object body, TypeToken itemType) { + return serdes.deserialize(JSON.serialize(body), itemType); + } + + private static Object toJsonValue(SerDes serdes, Object value) { + return JSON.deserialize(serdes.serialize(value), OBJECT_TYPE); + } + + private static Map resultEntry(String itemId, Object output) { + var entry = new LinkedHashMap(); + entry.put("itemIdentifier", itemId); + entry.put("output", output); + return entry; + } + + private static Map errorEntry(String itemId, String errorType, String errorMessage) { + var error = new LinkedHashMap(); + error.put("errorType", errorType); + error.put("errorMessage", errorMessage); + var entry = new LinkedHashMap(); + entry.put("itemIdentifier", itemId); + entry.put("error", error); + return entry; + } + + private static Map itemResponse( + ResponseMode report, List> results, List> failures) { + var response = new LinkedHashMap(); + if (report == ResponseMode.REPORT_FAILED_ITEMS) { + response.put("batchItemFailures", failures); + } else { + response.put("batchItemResults", results); + response.put("batchItemFailures", failures); + } + return response; + } +} 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..79a00cdc8 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/DurableContext.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/DurableContext.java @@ -9,6 +9,9 @@ import java.util.function.Function; import java.util.function.Supplier; import software.amazon.lambda.durable.config.CallbackConfig; +import software.amazon.lambda.durable.config.DistributedMapConfig; +import software.amazon.lambda.durable.config.DistributedMapProcessor; +import software.amazon.lambda.durable.config.DistributedMapSource; import software.amazon.lambda.durable.config.InvokeConfig; import software.amazon.lambda.durable.config.MapConfig; import software.amazon.lambda.durable.config.ParallelConfig; @@ -18,6 +21,8 @@ import software.amazon.lambda.durable.config.WaitForConditionConfig; import software.amazon.lambda.durable.config.WithRetryConfig; import software.amazon.lambda.durable.context.BaseContext; +import software.amazon.lambda.durable.model.DistributedMapResult; +import software.amazon.lambda.durable.model.DistributedMapSummary; import software.amazon.lambda.durable.model.MapResult; import software.amazon.lambda.durable.model.WaitForConditionResult; @@ -543,6 +548,142 @@ default ParallelDurableFuture parallel(String name) { */ ParallelDurableFuture parallel(String name, ParallelConfig config); + // =============== distributedMap ================ + + /** Runs a distributed map with default config, blocking until the run reaches a terminal state. */ + default DistributedMapSummary distributedMap( + String name, DistributedMapSource source, DistributedMapProcessor processor, int maxConcurrency) { + return distributedMapAsync( + name, source, processor, maxConcurrency, DistributedMapConfig.builder().build()) + .get(); + } + + /** Runs a distributed map, blocking until the run reaches a terminal state. */ + default DistributedMapSummary distributedMap( + String name, + DistributedMapSource source, + DistributedMapProcessor processor, + int maxConcurrency, + DistributedMapConfig config) { + return distributedMapAsync(name, source, processor, maxConcurrency, config) + .get(); + } + + /** Runs a distributed map that collects per-item results with a TypeToken, blocking until the run reaches a terminal state. */ + default DistributedMapResult distributedMap( + String name, + DistributedMapSource source, + DistributedMapProcessor processor, + int maxConcurrency, + TypeToken resultType, + DistributedMapConfig config) { + return distributedMapAsync(name, source, processor, maxConcurrency, resultType, config) + .get(); + } + + /** Runs a distributed map that collects per-item results with a result Class, blocking until the run reaches a terminal state. */ + default DistributedMapResult distributedMap( + String name, + DistributedMapSource source, + DistributedMapProcessor processor, + int maxConcurrency, + Class resultType, + DistributedMapConfig config) { + return distributedMapAsync(name, source, processor, maxConcurrency, TypeToken.get(resultType), config) + .get(); + } + + /** Runs a distributed map that collects per-item results with a result Class and default config, blocking until the run reaches a terminal state. */ + default DistributedMapResult distributedMap( + String name, + DistributedMapSource source, + DistributedMapProcessor processor, + int maxConcurrency, + Class resultType) { + return distributedMapAsync( + name, + source, + processor, + maxConcurrency, + TypeToken.get(resultType), + DistributedMapConfig.builder().build()) + .get(); + } + + /** Runs a distributed map that collects per-item results with a TypeToken and default config, blocking until the run reaches a terminal state. */ + default DistributedMapResult distributedMap( + String name, + DistributedMapSource source, + DistributedMapProcessor processor, + int maxConcurrency, + TypeToken resultType) { + return distributedMapAsync( + name, source, processor, maxConcurrency, resultType, DistributedMapConfig.builder().build()) + .get(); + } + + /** Asynchronously runs a distributed map with default config. */ + default DurableFuture distributedMapAsync( + String name, DistributedMapSource source, DistributedMapProcessor processor, int maxConcurrency) { + return distributedMapAsync( + name, source, processor, maxConcurrency, DistributedMapConfig.builder().build()); + } + + /** Asynchronously runs a distributed map. */ + DurableFuture distributedMapAsync( + String name, + DistributedMapSource source, + DistributedMapProcessor processor, + int maxConcurrency, + DistributedMapConfig config); + + /** Asynchronously runs a distributed map that collects per-item results with a TypeToken. */ + DurableFuture> distributedMapAsync( + String name, + DistributedMapSource source, + DistributedMapProcessor processor, + int maxConcurrency, + TypeToken resultType, + DistributedMapConfig config); + + /** Asynchronously runs a distributed map that collects per-item results with a result Class. */ + default DurableFuture> distributedMapAsync( + String name, + DistributedMapSource source, + DistributedMapProcessor processor, + int maxConcurrency, + Class resultType, + DistributedMapConfig config) { + return distributedMapAsync(name, source, processor, maxConcurrency, TypeToken.get(resultType), config); + } + + /** Asynchronously runs a distributed map that collects per-item results with a result Class and default config. */ + default DurableFuture> distributedMapAsync( + String name, + DistributedMapSource source, + DistributedMapProcessor processor, + int maxConcurrency, + Class resultType) { + return distributedMapAsync( + name, + source, + processor, + maxConcurrency, + TypeToken.get(resultType), + DistributedMapConfig.builder().build()); + } + + /** Asynchronously runs a distributed map that collects per-item results with a TypeToken and default config. */ + default DurableFuture> distributedMapAsync( + String name, + DistributedMapSource source, + DistributedMapProcessor processor, + int maxConcurrency, + TypeToken resultType) { + return distributedMapAsync( + name, source, processor, maxConcurrency, resultType, DistributedMapConfig.builder().build()); + } + /** * Executes a submitter function and waits for an external callback, blocking until the callback completes. * diff --git a/sdk/src/main/java/software/amazon/lambda/durable/config/CsvDelimiter.java b/sdk/src/main/java/software/amazon/lambda/durable/config/CsvDelimiter.java new file mode 100644 index 000000000..df6d4a355 --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/config/CsvDelimiter.java @@ -0,0 +1,23 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.config; + +/** Delimiter for CSV distributed map sources. */ +public enum CsvDelimiter { + COMMA("COMMA"), + PIPE("PIPE"), + SEMICOLON("SEMICOLON"), + SPACE("SPACE"), + TAB("TAB"); + + private final String value; + + CsvDelimiter(String value) { + this.value = value; + } + + /** Returns the wire-format string value. */ + public String getValue() { + return value; + } +} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/config/CsvFormat.java b/sdk/src/main/java/software/amazon/lambda/durable/config/CsvFormat.java new file mode 100644 index 000000000..61d9e2cd3 --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/config/CsvFormat.java @@ -0,0 +1,45 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.config; + +import java.util.List; +import software.amazon.lambda.durable.util.DistributedMapValidation; + +/** CSV parsing options for a distributed map S3 source. */ +public record CsvFormat(HeaderLocation headerLocation, List columns, CsvDelimiter delimiter) { + + /** Where the column header row comes from. */ + public enum HeaderLocation { + FIRST_ROW, + GIVEN + } + + public CsvFormat { + columns = columns != null ? List.copyOf(columns) : null; + delimiter = delimiter != null ? delimiter : CsvDelimiter.COMMA; + } + + /** + * File has its own header row. Column names are used only for compile-time typing and are not sent on the wire. + */ + public static CsvFormat expectedColumns(List columns) { + DistributedMapValidation.validateColumns("expectedColumns", columns); + return new CsvFormat(HeaderLocation.FIRST_ROW, columns, CsvDelimiter.COMMA); + } + + /** File has no header row. The given names are the columns, sent on the wire, and the first row is data. */ + public static CsvFormat headers(List headers) { + DistributedMapValidation.validateColumns("headers", headers); + return new CsvFormat(HeaderLocation.GIVEN, headers, CsvDelimiter.COMMA); + } + + /** File has its own header row and no column names are declared. */ + public static CsvFormat firstRow() { + return new CsvFormat(HeaderLocation.FIRST_ROW, null, CsvDelimiter.COMMA); + } + + /** Returns a copy with the given delimiter. */ + public CsvFormat withDelimiter(CsvDelimiter delimiter) { + return new CsvFormat(headerLocation, columns, delimiter); + } +} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/config/DistributedMapCompletionConfig.java b/sdk/src/main/java/software/amazon/lambda/durable/config/DistributedMapCompletionConfig.java new file mode 100644 index 000000000..ee5bd864b --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/config/DistributedMapCompletionConfig.java @@ -0,0 +1,45 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.config; + +/** Failure-tolerance configuration for a distributed map run. */ +public record DistributedMapCompletionConfig( + Integer toleratedFailureCount, Double toleratedFailurePercentage, Integer minimumSampleSize) { + + public DistributedMapCompletionConfig { + if (toleratedFailureCount != null && toleratedFailurePercentage != null) { + throw new IllegalArgumentException( + "toleratedFailureCount and toleratedFailurePercentage are mutually exclusive"); + } + if (minimumSampleSize != null && toleratedFailurePercentage == null) { + throw new IllegalArgumentException("minimumSampleSize is only valid with toleratedFailurePercentage"); + } + if (toleratedFailureCount != null && toleratedFailureCount < 0) { + throw new IllegalArgumentException( + "toleratedFailureCount must be non-negative, got: " + toleratedFailureCount); + } + if (toleratedFailurePercentage != null + && (toleratedFailurePercentage < 0 || toleratedFailurePercentage > 100)) { + throw new IllegalArgumentException( + "toleratedFailurePercentage must be between 0 and 100, got: " + toleratedFailurePercentage); + } + if (minimumSampleSize != null && minimumSampleSize < 1) { + throw new IllegalArgumentException("minimumSampleSize must be at least 1, got: " + minimumSampleSize); + } + } + + /** Abort once this many items have permanently failed. */ + public static DistributedMapCompletionConfig failureCount(int count) { + return new DistributedMapCompletionConfig(count, null, null); + } + + /** Abort once the failure rate exceeds this percentage (0 to 100). */ + public static DistributedMapCompletionConfig failurePercentage(double percentage) { + return new DistributedMapCompletionConfig(null, percentage, null); + } + + /** Abort once the failure rate exceeds this percentage (0 to 100), gated by a minimum sample size. */ + public static DistributedMapCompletionConfig failurePercentage(double percentage, int minimumSampleSize) { + return new DistributedMapCompletionConfig(null, percentage, minimumSampleSize); + } +} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/config/DistributedMapConfig.java b/sdk/src/main/java/software/amazon/lambda/durable/config/DistributedMapConfig.java new file mode 100644 index 000000000..0ea953c1e --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/config/DistributedMapConfig.java @@ -0,0 +1,91 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.config; + +import java.time.Duration; +import software.amazon.lambda.durable.serde.SerDes; + +/** Optional configuration for a distributed map run. */ +public class DistributedMapConfig { + private static final long MAX_TIMEOUT_SECONDS = 7776000; // 90 days + + private final DistributedMapDestinationConfig destination; + private final DistributedMapCompletionConfig completionConfig; + private final Duration timeout; + private final SerDes resultSerDes; + + private DistributedMapConfig(Builder builder) { + if (builder.timeout != null + && (builder.timeout.toSeconds() <= 0 || builder.timeout.toSeconds() > MAX_TIMEOUT_SECONDS)) { + throw new IllegalArgumentException( + "timeout must be positive and at most 90 days, got: " + builder.timeout.toSeconds() + "s"); + } + this.destination = builder.destination; + this.completionConfig = builder.completionConfig; + this.timeout = builder.timeout; + this.resultSerDes = builder.resultSerDes; + } + + public DistributedMapDestinationConfig destination() { + return destination; + } + + public DistributedMapCompletionConfig completionConfig() { + return completionConfig; + } + + public Duration timeout() { + return timeout; + } + + public SerDes resultSerDes() { + return resultSerDes; + } + + public static Builder builder() { + return new Builder(); + } + + public Builder toBuilder() { + var builder = new Builder(); + builder.destination = destination; + builder.completionConfig = completionConfig; + builder.timeout = timeout; + builder.resultSerDes = resultSerDes; + return builder; + } + + /** Builder for DistributedMapConfig. */ + public static class Builder { + private DistributedMapDestinationConfig destination; + private DistributedMapCompletionConfig completionConfig; + private Duration timeout; + private SerDes resultSerDes; + + private Builder() {} + + public Builder destination(DistributedMapDestinationConfig destination) { + this.destination = destination; + return this; + } + + public Builder completionConfig(DistributedMapCompletionConfig completionConfig) { + this.completionConfig = completionConfig; + return this; + } + + public Builder timeout(Duration timeout) { + this.timeout = timeout; + return this; + } + + public Builder resultSerDes(SerDes resultSerDes) { + this.resultSerDes = resultSerDes; + return this; + } + + public DistributedMapConfig build() { + return new DistributedMapConfig(this); + } + } +} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/config/DistributedMapDestination.java b/sdk/src/main/java/software/amazon/lambda/durable/config/DistributedMapDestination.java new file mode 100644 index 000000000..7a9dd1bb1 --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/config/DistributedMapDestination.java @@ -0,0 +1,44 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.config; + +import software.amazon.lambda.durable.util.DistributedMapValidation; + +/** Destination factories for a distributed map run. */ +public final class DistributedMapDestination { + private DistributedMapDestination() {} + + /** Route succeeded item records to an S3 prefix (defaults to include output only). */ + public static SuccessDestination successes(String prefixUri) { + return successes(prefixUri, false, true, null); + } + + /** Route succeeded item records to an S3 prefix. */ + public static SuccessDestination successes( + String prefixUri, boolean includeInput, boolean includeOutput, String expectedBucketOwner) { + var parsed = DistributedMapValidation.parseS3Uri(prefixUri); + return new SuccessDestination( + parsed.bucket(), + parsed.path() != null ? parsed.path() : "", + includeInput, + includeOutput, + expectedBucketOwner); + } + + /** Route permanently-failed item records to an S3 prefix (defaults to include input and error). */ + public static FailureDestination failures(String prefixUri) { + return failures(prefixUri, true, true, null); + } + + /** Route permanently-failed item records to an S3 prefix. */ + public static FailureDestination failures( + String prefixUri, boolean includeInput, boolean includeError, String expectedBucketOwner) { + var parsed = DistributedMapValidation.parseS3Uri(prefixUri); + return new FailureDestination( + parsed.bucket(), + parsed.path() != null ? parsed.path() : "", + includeInput, + includeError, + expectedBucketOwner); + } +} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/config/DistributedMapDestinationConfig.java b/sdk/src/main/java/software/amazon/lambda/durable/config/DistributedMapDestinationConfig.java new file mode 100644 index 000000000..4ce47e60b --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/config/DistributedMapDestinationConfig.java @@ -0,0 +1,22 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.config; + +/** Destination routing for distributed map run results. */ +public record DistributedMapDestinationConfig(SuccessDestination onSuccess, FailureDestination onFailure) { + + /** Routes only succeeded item records. */ + public static DistributedMapDestinationConfig onSuccess(SuccessDestination onSuccess) { + return new DistributedMapDestinationConfig(onSuccess, null); + } + + /** Routes only failed item records. */ + public static DistributedMapDestinationConfig onFailure(FailureDestination onFailure) { + return new DistributedMapDestinationConfig(null, onFailure); + } + + /** Routes both succeeded and failed item records. */ + public static DistributedMapDestinationConfig of(SuccessDestination onSuccess, FailureDestination onFailure) { + return new DistributedMapDestinationConfig(onSuccess, onFailure); + } +} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/config/DistributedMapProcessor.java b/sdk/src/main/java/software/amazon/lambda/durable/config/DistributedMapProcessor.java new file mode 100644 index 000000000..db38c7e94 --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/config/DistributedMapProcessor.java @@ -0,0 +1,123 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.config; + +import software.amazon.lambda.durable.util.DistributedMapValidation; + +/** Processor configuration for a distributed map run. */ +public class DistributedMapProcessor { + + /** Per-batch response mode reported by the processor. */ + public enum ResponseMode { + REPORT_BATCH_OUTCOME(null), + REPORT_FAILED_ITEMS("REPORT_BATCH_ITEM_FAILURES"), + REPORT_ITEM_RESULTS("REPORT_BATCH_ITEM_RESULTS"); + + private final String value; + + ResponseMode(String value) { + this.value = value; + } + + /** Returns the wire functionResponseTypes value, or null for the batch-outcome mode. */ + public String getValue() { + return value; + } + } + + private static final int MIN_BATCH_SIZE = 1; + private static final int MAX_BATCH_SIZE = 10000; + private static final int MAX_NAME_PREFIX_LENGTH = 36; + + private final String functionName; + private final ResponseMode responseMode; + private final Integer batchSize; + private final ProcessorRetryConfig retryConfig; + private final String durableExecutionNamePrefix; + + private DistributedMapProcessor(Builder builder) { + DistributedMapValidation.validateFunctionName(builder.functionName); + if (builder.batchSize != null && (builder.batchSize < MIN_BATCH_SIZE || builder.batchSize > MAX_BATCH_SIZE)) { + throw new IllegalArgumentException("batchSize must be between 1 and 10000, got: " + builder.batchSize); + } + if (builder.durableExecutionNamePrefix != null + && (builder.durableExecutionNamePrefix.isEmpty() + || builder.durableExecutionNamePrefix.length() > MAX_NAME_PREFIX_LENGTH)) { + throw new IllegalArgumentException("durableExecutionNamePrefix must be between 1 and 36 characters, got: " + + builder.durableExecutionNamePrefix.length()); + } + this.functionName = builder.functionName; + this.responseMode = builder.responseMode; + this.batchSize = builder.batchSize; + this.retryConfig = builder.retryConfig; + this.durableExecutionNamePrefix = builder.durableExecutionNamePrefix; + } + + public String functionName() { + return functionName; + } + + public ResponseMode responseMode() { + return responseMode; + } + + public Integer batchSize() { + return batchSize; + } + + public ProcessorRetryConfig retryConfig() { + return retryConfig; + } + + public String durableExecutionNamePrefix() { + return durableExecutionNamePrefix; + } + + /** Processor that reports a single pass/fail outcome for the whole batch, with no per-item results. */ + public static Builder reportBatchOutcome(String functionName) { + return new Builder(functionName, ResponseMode.REPORT_BATCH_OUTCOME); + } + + /** Processor that reports the ids of failed items, with all others marked succeeded. */ + public static Builder reportFailedItems(String functionName) { + return new Builder(functionName, ResponseMode.REPORT_FAILED_ITEMS); + } + + /** Processor that reports the results (output or error) for every item. */ + public static Builder reportItemResults(String functionName) { + return new Builder(functionName, ResponseMode.REPORT_ITEM_RESULTS); + } + + /** Builder for DistributedMapProcessor. */ + public static class Builder { + private final String functionName; + private final ResponseMode responseMode; + private Integer batchSize; + private ProcessorRetryConfig retryConfig; + private String durableExecutionNamePrefix; + + private Builder(String functionName, ResponseMode responseMode) { + this.functionName = functionName; + this.responseMode = responseMode; + } + + public Builder batchSize(int batchSize) { + this.batchSize = batchSize; + return this; + } + + public Builder retryConfig(ProcessorRetryConfig retryConfig) { + this.retryConfig = retryConfig; + return this; + } + + public Builder durableExecutionNamePrefix(String durableExecutionNamePrefix) { + this.durableExecutionNamePrefix = durableExecutionNamePrefix; + return this; + } + + public DistributedMapProcessor build() { + return new DistributedMapProcessor(this); + } + } +} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/config/DistributedMapSource.java b/sdk/src/main/java/software/amazon/lambda/durable/config/DistributedMapSource.java new file mode 100644 index 000000000..3d3607d4c --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/config/DistributedMapSource.java @@ -0,0 +1,212 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.config; + +import java.util.Collection; +import java.util.List; +import software.amazon.lambda.durable.serde.SerDes; +import software.amazon.lambda.durable.util.DistributedMapValidation; +import software.amazon.lambda.durable.util.ParameterValidator; + +/** + * Source of items for a distributed map run. + * + * @param the item type produced by the source + */ +public class DistributedMapSource { + + /** The kind of source backing a distributed map. */ + public enum SourceType { + INLINE, + S3, + READER_FUNCTION + } + + /** How the bytes of an S3 object are split into items. */ + public enum Format { + JSON_LINES, + JSON_ARRAY, + CSV + } + + /** Whether the backend flattens object contents into items or treats each object as one item. */ + public enum Transform { + NONE, + LOAD_AND_FLATTEN + } + + private final SourceType sourceType; + private final Long maxItems; + private final List inlineItems; + private final SerDes inlineSerDes; + private final S3SourceConfig s3; + private final ReaderSourceConfig reader; + + private DistributedMapSource( + SourceType sourceType, + Long maxItems, + List inlineItems, + SerDes inlineSerDes, + S3SourceConfig s3, + ReaderSourceConfig reader) { + if (maxItems != null && maxItems < 1) { + throw new IllegalArgumentException("maxItems must be at least 1, got: " + maxItems); + } + this.sourceType = sourceType; + this.maxItems = maxItems; + this.inlineItems = inlineItems != null ? List.copyOf(inlineItems) : null; + this.inlineSerDes = inlineSerDes; + this.s3 = s3; + this.reader = reader; + } + + public SourceType sourceType() { + return sourceType; + } + + public Long maxItems() { + return maxItems; + } + + public List inlineItems() { + return inlineItems; + } + + public SerDes inlineSerDes() { + return inlineSerDes; + } + + public S3SourceConfig s3() { + return s3; + } + + public ReaderSourceConfig reader() { + return reader; + } + + /** An in-memory list of items embedded in the start checkpoint. */ + public static DistributedMapSource inline(Collection items) { + return inline(items, null); + } + + /** An in-memory list of items embedded in the start checkpoint, serialized with the given SerDes. */ + public static DistributedMapSource inline(Collection items, SerDes serDes) { + ParameterValidator.validateOrderedCollection(items); + return new DistributedMapSource<>(SourceType.INLINE, null, List.copyOf(items), serDes, null, null); + } + + /** Read a single S3 object, treating each line as an item. */ + public static DistributedMapSource s3JsonLines(String uri) { + var parsed = DistributedMapValidation.parseS3Uri(uri); + if (parsed.path() == null) { + throw new IllegalArgumentException("s3JsonLines requires an S3 object key"); + } + return s3(new S3SourceConfig(parsed.bucket(), parsed.path(), null, null, Format.JSON_LINES, null, null)); + } + + /** Read a single S3 object holding a JSON array, treating each element as an item. */ + public static DistributedMapSource s3JsonArray(String uri) { + var parsed = DistributedMapValidation.parseS3Uri(uri); + if (parsed.path() == null) { + throw new IllegalArgumentException("s3JsonArray requires an S3 object key"); + } + return s3(new S3SourceConfig(parsed.bucket(), parsed.path(), null, null, Format.JSON_ARRAY, null, null)); + } + + /** Read a single S3 object, treating each record as an item. */ + public static DistributedMapSource s3Csv(String uri, CsvFormat format) { + var parsed = DistributedMapValidation.parseS3Uri(uri); + if (parsed.path() == null) { + throw new IllegalArgumentException("s3Csv requires an S3 object key"); + } + return s3(new S3SourceConfig(parsed.bucket(), parsed.path(), null, null, Format.CSV, format, null)); + } + + /** Read each object under a prefix as one item (object contents are not read). */ + public static DistributedMapSource s3Objects(String prefixUri) { + var parsed = DistributedMapValidation.parseS3Uri(prefixUri); + return s3(new S3SourceConfig(parsed.bucket(), null, prefixOrEmpty(parsed.path()), Transform.NONE, null, null, null)); + } + + /** Read a prefix, flattening each object's lines into items. */ + public static DistributedMapSource s3FlattenedJsonLines(String prefixUri) { + var parsed = DistributedMapValidation.parseS3Uri(prefixUri); + return s3(new S3SourceConfig( + parsed.bucket(), null, prefixOrEmpty(parsed.path()), Transform.LOAD_AND_FLATTEN, Format.JSON_LINES, null, null)); + } + + /** Read a prefix, flattening each object's JSON array elements into items. */ + public static DistributedMapSource s3FlattenedJsonArray(String prefixUri) { + var parsed = DistributedMapValidation.parseS3Uri(prefixUri); + return s3(new S3SourceConfig( + parsed.bucket(), null, prefixOrEmpty(parsed.path()), Transform.LOAD_AND_FLATTEN, Format.JSON_ARRAY, null, null)); + } + + /** Read a prefix, flattening each object's records into items. */ + public static DistributedMapSource s3FlattenedCsv(String prefixUri, CsvFormat format) { + var parsed = DistributedMapValidation.parseS3Uri(prefixUri); + return s3(new S3SourceConfig( + parsed.bucket(), null, prefixOrEmpty(parsed.path()), Transform.LOAD_AND_FLATTEN, Format.CSV, format, null)); + } + + /** Page items from a customer-supplied reader Lambda function. */ + public static DistributedMapSource reader(String functionName) { + return reader(functionName, null, null); + } + + /** Page items from a customer-supplied reader Lambda function with an initial state. */ + public static DistributedMapSource reader(String functionName, S initialState, SerDes stateSerDes) { + return new DistributedMapSource<>( + SourceType.READER_FUNCTION, + null, + null, + null, + null, + new ReaderSourceConfig(functionName, initialState, stateSerDes)); + } + + /** Returns a copy limiting the total number of items read. */ + public DistributedMapSource maxItemsToRead(long maxItems) { + return new DistributedMapSource<>(sourceType, maxItems, inlineItems, inlineSerDes, s3, reader); + } + + /** Returns a copy asserting the expected S3 bucket owner (S3 sources only). */ + public DistributedMapSource expectedBucketOwner(String accountId) { + if (s3 == null) { + throw new IllegalStateException("expectedBucketOwner only applies to S3 sources"); + } + DistributedMapValidation.validateBucketOwner(accountId); + return new DistributedMapSource<>( + sourceType, maxItems, inlineItems, inlineSerDes, s3.withExpectedBucketOwner(accountId), reader); + } + + private static DistributedMapSource s3(S3SourceConfig s3) { + return new DistributedMapSource<>(SourceType.S3, null, null, null, s3, null); + } + + private static String prefixOrEmpty(String path) { + return path != null ? path : ""; + } + + /** Resolved S3 source configuration. */ + public record S3SourceConfig( + String bucket, + String key, + String prefix, + Transform transform, + Format format, + CsvFormat csvFormat, + String expectedBucketOwner) { + + S3SourceConfig withExpectedBucketOwner(String owner) { + return new S3SourceConfig(bucket, key, prefix, transform, format, csvFormat, owner); + } + } + + /** Resolved reader-function source configuration. */ + public record ReaderSourceConfig(String functionName, Object initialState, SerDes stateSerDes) { + public ReaderSourceConfig { + DistributedMapValidation.validateFunctionName(functionName); + } + } +} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/config/FailureDestination.java b/sdk/src/main/java/software/amazon/lambda/durable/config/FailureDestination.java new file mode 100644 index 000000000..87a9aa822 --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/config/FailureDestination.java @@ -0,0 +1,17 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.config; + +import software.amazon.lambda.durable.util.DistributedMapValidation; + +/** S3 destination for permanently-failed distributed map items. */ +public record FailureDestination( + String bucket, String prefix, boolean includeInput, boolean includeError, String expectedBucketOwner) { + + public FailureDestination { + DistributedMapValidation.validateBucketOwner(expectedBucketOwner); + if (!includeInput && !includeError) { + throw new IllegalArgumentException("failure destination must include at least one of input or error"); + } + } +} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/config/ProcessorRetryConfig.java b/sdk/src/main/java/software/amazon/lambda/durable/config/ProcessorRetryConfig.java new file mode 100644 index 000000000..4fa361fce --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/config/ProcessorRetryConfig.java @@ -0,0 +1,76 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.config; + +import java.time.Duration; + +/** Retry configuration for a distributed map processor. */ +public class ProcessorRetryConfig { + /** Sentinel for unbounded retry attempts (wire value -1). */ + public static final int UNLIMITED = -1; + + private static final long MIN_RETRY_DURATION_SECONDS = 60; + private static final long MAX_RETRY_DURATION_SECONDS = 21600; + + private final Integer maxRetryAttempts; + private final Duration maxRetryDuration; + + private ProcessorRetryConfig(Builder builder) { + this.maxRetryAttempts = builder.maxRetryAttempts; + this.maxRetryDuration = builder.maxRetryDuration; + } + + /** Returns the max retry attempts, UNLIMITED for unbounded, or null for the default. */ + public Integer maxRetryAttempts() { + return maxRetryAttempts; + } + + /** Returns the cumulative retry duration budget, or null for the default. */ + public Duration maxRetryDuration() { + return maxRetryDuration; + } + + public static Builder builder() { + return new Builder(); + } + + public Builder toBuilder() { + var builder = new Builder(); + builder.maxRetryAttempts = maxRetryAttempts; + builder.maxRetryDuration = maxRetryDuration; + return builder; + } + + /** Builder for ProcessorRetryConfig. */ + public static class Builder { + private Integer maxRetryAttempts; + private Duration maxRetryDuration; + + private Builder() {} + + public Builder maxRetryAttempts(int maxRetryAttempts) { + if (maxRetryAttempts < 0 && maxRetryAttempts != UNLIMITED) { + throw new IllegalArgumentException( + "maxRetryAttempts must be non-negative or ProcessorRetryConfig.UNLIMITED, got: " + + maxRetryAttempts); + } + this.maxRetryAttempts = maxRetryAttempts; + return this; + } + + public Builder maxRetryDuration(Duration maxRetryDuration) { + if (maxRetryDuration != null + && (maxRetryDuration.toSeconds() < MIN_RETRY_DURATION_SECONDS + || maxRetryDuration.toSeconds() > MAX_RETRY_DURATION_SECONDS)) { + throw new IllegalArgumentException("maxRetryDuration must be between 1 minute and 6 hours, got: " + + maxRetryDuration.toSeconds() + "s"); + } + this.maxRetryDuration = maxRetryDuration; + return this; + } + + public ProcessorRetryConfig build() { + return new ProcessorRetryConfig(this); + } + } +} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/config/SuccessDestination.java b/sdk/src/main/java/software/amazon/lambda/durable/config/SuccessDestination.java new file mode 100644 index 000000000..42a31cd0f --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/config/SuccessDestination.java @@ -0,0 +1,17 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.config; + +import software.amazon.lambda.durable.util.DistributedMapValidation; + +/** S3 destination for succeeded distributed map items. */ +public record SuccessDestination( + String bucket, String prefix, boolean includeInput, boolean includeOutput, String expectedBucketOwner) { + + public SuccessDestination { + DistributedMapValidation.validateBucketOwner(expectedBucketOwner); + if (!includeInput && !includeOutput) { + throw new IllegalArgumentException("success destination must include at least one of input or output"); + } + } +} 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 0c79165ec..69a45cc88 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 @@ -3,6 +3,7 @@ package software.amazon.lambda.durable.context; import com.amazonaws.services.lambda.runtime.Context; +import software.amazon.awssdk.services.lambda.model.DistributedMapDetails; import java.time.Duration; import java.util.Collection; import java.util.List; @@ -18,6 +19,9 @@ import software.amazon.lambda.durable.StepContext; import software.amazon.lambda.durable.TypeToken; import software.amazon.lambda.durable.config.CallbackConfig; +import software.amazon.lambda.durable.config.DistributedMapConfig; +import software.amazon.lambda.durable.config.DistributedMapProcessor; +import software.amazon.lambda.durable.config.DistributedMapSource; import software.amazon.lambda.durable.config.InvokeConfig; import software.amazon.lambda.durable.config.MapConfig; import software.amazon.lambda.durable.config.ParallelConfig; @@ -31,12 +35,16 @@ import software.amazon.lambda.durable.execution.OperationIdGenerator; import software.amazon.lambda.durable.execution.SuspendExecutionException; import software.amazon.lambda.durable.execution.ThreadType; +import software.amazon.lambda.durable.model.DistributedMapResult; +import software.amazon.lambda.durable.model.DistributedMapSummary; import software.amazon.lambda.durable.model.MapResult; import software.amazon.lambda.durable.model.OperationIdentifier; import software.amazon.lambda.durable.model.OperationSubType; import software.amazon.lambda.durable.model.WaitForConditionResult; import software.amazon.lambda.durable.operation.CallbackOperation; import software.amazon.lambda.durable.operation.ChildContextOperation; +import software.amazon.lambda.durable.operation.DistributedMapOperation; +import software.amazon.lambda.durable.operation.DistributedMapWire; import software.amazon.lambda.durable.operation.InvokeOperation; import software.amazon.lambda.durable.operation.MapOperation; import software.amazon.lambda.durable.operation.ParallelOperation; @@ -371,6 +379,85 @@ public DurableFuture waitForConditionAsync( return operation; } + // =============== distributedMap ================ + + @Override + public DurableFuture distributedMapAsync( + String name, + DistributedMapSource source, + DistributedMapProcessor processor, + int maxConcurrency, + DistributedMapConfig config) { + return startDistributedMap( + name, + source, + processor, + maxConcurrency, + config, + false, + TypeToken.get(DistributedMapSummary.class), + DistributedMapWire::toSummary); + } + + @Override + public DurableFuture> distributedMapAsync( + String name, + DistributedMapSource source, + DistributedMapProcessor processor, + int maxConcurrency, + TypeToken resultType, + DistributedMapConfig config) { + Objects.requireNonNull(resultType, "resultType cannot be null"); + Objects.requireNonNull(config, "config cannot be null"); + var resultSerDes = + config.resultSerDes() != null ? config.resultSerDes() : getDurableConfig().getSerDes(); + @SuppressWarnings({"unchecked", "rawtypes"}) + TypeToken> token = (TypeToken) TypeToken.get(DistributedMapResult.class); + return startDistributedMap( + name, + source, + processor, + maxConcurrency, + config, + true, + token, + details -> DistributedMapWire.toResult(details, resultType, resultSerDes)); + } + + private DurableFuture startDistributedMap( + String name, + DistributedMapSource source, + DistributedMapProcessor processor, + int maxConcurrency, + DistributedMapConfig config, + boolean collectResults, + TypeToken resultType, + Function resultBuilder) { + Objects.requireNonNull(source, "source cannot be null"); + Objects.requireNonNull(processor, "processor cannot be null"); + Objects.requireNonNull(config, "config cannot be null"); + ParameterValidator.validateOperationName(name); + if (maxConcurrency <= 0) { + throw new IllegalArgumentException("maxConcurrency must be greater than zero, got: " + maxConcurrency); + } + if (!collectResults && config.resultSerDes() != null) { + throw new IllegalArgumentException( + "resultSerDes is set but this distributedMap call does not collect results, use a resultType overload"); + } + var options = DistributedMapWire.toOptions( + source, processor, maxConcurrency, config, getDurableConfig().getSerDes(), collectResults); + var operationId = nextOperationId(); + var operation = new DistributedMapOperation<>( + OperationIdentifier.of(operationId, name, OperationSubType.DISTRIBUTED_MAP), + options, + resultType, + getDurableConfig().getSerDes(), + resultBuilder, + this); + operation.execute(); + return operation; + } + // =============== withRetry ================ private static final Duration DEFAULT_BACKOFF_DELAY = Duration.ofSeconds(1); diff --git a/sdk/src/main/java/software/amazon/lambda/durable/exception/DistributedMapError.java b/sdk/src/main/java/software/amazon/lambda/durable/exception/DistributedMapError.java new file mode 100644 index 000000000..49d90359c --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/exception/DistributedMapError.java @@ -0,0 +1,93 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.exception; + +import software.amazon.lambda.durable.model.DistributedMapCompletionReason; +import software.amazon.lambda.durable.model.DistributedMapItemError; +import software.amazon.lambda.durable.model.DistributedMapStatus; + +/** + * Run outcome error, thrown by throwIfError when a distributed map run did not fully succeed. + * It reports an aggregate run result and is not a single operation failure, so it stays out of the DurableOperationException hierarchy. + * See DistributedMapException for a failure of the operation itself. + */ +public class DistributedMapError extends DurableExecutionException { + private final DistributedMapStatus status; + private final DistributedMapCompletionReason completionReason; + private final long failureCount; + + private DistributedMapError( + DistributedMapStatus status, + DistributedMapCompletionReason completionReason, + long failureCount, + String message) { + super(message); + this.status = status; + this.completionReason = completionReason; + this.failureCount = failureCount; + } + + /** Run-level failure describing the run status, reason, and failure count. */ + public static DistributedMapError runLevel( + DistributedMapStatus status, + DistributedMapCompletionReason completionReason, + long failureCount, + String completionDetails) { + return new DistributedMapError( + status, + completionReason, + failureCount, + runMessage(status, completionReason, failureCount, completionDetails)); + } + + /** Item-level failure surfacing the first failed item's error. */ + public static DistributedMapError itemLevel( + DistributedMapStatus status, + DistributedMapCompletionReason completionReason, + long failureCount, + DistributedMapItemError error, + String itemId) { + return new DistributedMapError(status, completionReason, failureCount, itemMessage(error, itemId)); + } + + /** Returns the terminal run status. */ + public DistributedMapStatus status() { + return status; + } + + /** Returns the completion reason. */ + public DistributedMapCompletionReason completionReason() { + return completionReason; + } + + /** Returns the count of failed items. */ + public long failureCount() { + return failureCount; + } + + private static String runMessage( + DistributedMapStatus status, + DistributedMapCompletionReason completionReason, + long failureCount, + String completionDetails) { + var message = new StringBuilder("Distributed map run ") + .append(status) + .append(" (") + .append(completionReason) + .append(")"); + if (failureCount > 0) { + message.append(", ").append(failureCount).append(" item(s) failed"); + } + if (completionDetails != null) { + message.append(": ").append(completionDetails); + } + return message.toString(); + } + + private static String itemMessage(DistributedMapItemError error, String itemId) { + if (error != null) { + return error.errorType() + ": " + error.errorMessage(); + } + return "Distributed map item " + itemId + " failed"; + } +} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/exception/DistributedMapException.java b/sdk/src/main/java/software/amazon/lambda/durable/exception/DistributedMapException.java new file mode 100644 index 000000000..d8ee6d20d --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/exception/DistributedMapException.java @@ -0,0 +1,18 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.exception; + +import software.amazon.awssdk.services.lambda.model.Operation; + +/** + * Operation failure, thrown when a distributed map operation itself terminal-fails (FAILED, TIMED_OUT, or STOPPED). + * Extends DurableOperationException. See DistributedMapError for the run outcome, which is an aggregate result rather than an operation failure. + */ +public class DistributedMapException extends DurableOperationException { + public DistributedMapException(Operation operation) { + super( + operation, + null, + "Distributed map operation " + operation.id() + " ended with status " + operation.statusAsString()); + } +} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/model/DistributedMapCompletionReason.java b/sdk/src/main/java/software/amazon/lambda/durable/model/DistributedMapCompletionReason.java new file mode 100644 index 000000000..17377cbf5 --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/model/DistributedMapCompletionReason.java @@ -0,0 +1,47 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.model; + +/** Reason a distributed map run reached its terminal status. */ +public enum DistributedMapCompletionReason { + ALL_COMPLETED("ALL_COMPLETED"), + ITEM_LIMIT_REACHED("ITEM_LIMIT_REACHED"), + STOPPED("STOPPED"), + TIMED_OUT("TIMED_OUT"), + FAILURE_TOLERANCE_EXCEEDED("FAILURE_TOLERANCE_EXCEEDED"), + SOURCE_FAILED("SOURCE_FAILED"), + DESTINATION_FAILED("DESTINATION_FAILED"), + INLINE_RESULT_LIMIT_EXCEEDED("INLINE_RESULT_LIMIT_EXCEEDED"), + INVALID_CONFIGURATION("INVALID_CONFIGURATION"), + QUOTA_EXCEEDED("QUOTA_EXCEEDED"), + KMS_ACCESS_DENIED("KMS_ACCESS_DENIED"), + INTERNAL_ERROR("INTERNAL_ERROR"), + /** A value the backend returned that this SDK version does not recognize. */ + UNKNOWN_TO_SDK_VERSION("UNKNOWN_TO_SDK_VERSION"); + + private final String value; + + DistributedMapCompletionReason(String value) { + this.value = value; + } + + /** Returns the wire-format string value. */ + public String getValue() { + return value; + } + + /** Returns the reason matching a wire-format string value, or UNKNOWN_TO_SDK_VERSION if unrecognized. */ + public static DistributedMapCompletionReason fromValue(String value) { + for (var reason : values()) { + if (reason.value.equals(value)) { + return reason; + } + } + return UNKNOWN_TO_SDK_VERSION; + } + + @Override + public String toString() { + return value; + } +} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/model/DistributedMapItemError.java b/sdk/src/main/java/software/amazon/lambda/durable/model/DistributedMapItemError.java new file mode 100644 index 000000000..4af34e166 --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/model/DistributedMapItemError.java @@ -0,0 +1,12 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.model; + +/** Error details for a failed distributed map item. */ +public record DistributedMapItemError(String errorType, String errorMessage) { + + /** Creates an item error from a throwable. */ + public static DistributedMapItemError of(Throwable e) { + return new DistributedMapItemError(e.getClass().getName(), e.getMessage()); + } +} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/model/DistributedMapResult.java b/sdk/src/main/java/software/amazon/lambda/durable/model/DistributedMapResult.java new file mode 100644 index 000000000..7a079eb5d --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/model/DistributedMapResult.java @@ -0,0 +1,112 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.model; + +import java.util.Collections; +import java.util.List; +import java.util.Objects; +import software.amazon.lambda.durable.exception.DistributedMapError; + +/** Outcome of a distributed map run with collected per-item results. */ +public record DistributedMapResult(DistributedMapSummary summary, List> items) { + + /** Applies a defensive copy and defaults. */ + public DistributedMapResult { + items = items != null ? List.copyOf(items) : Collections.emptyList(); + } + + /** Returns items that succeeded. */ + public List> succeeded() { + return items.stream() + .filter(item -> item.status() == DistributedMapResultItem.Status.SUCCEEDED) + .toList(); + } + + /** Returns items that failed. */ + public List> failed() { + return items.stream() + .filter(item -> item.status() == DistributedMapResultItem.Status.FAILED) + .toList(); + } + + /** Returns the outputs of succeeded items. */ + public List getResults() { + return succeeded().stream().map(DistributedMapResultItem::output).filter(Objects::nonNull).toList(); + } + + /** Returns the errors of failed items. */ + public List getErrors() { + return failed().stream().map(DistributedMapResultItem::error).filter(Objects::nonNull).toList(); + } + + /** Returns the terminal run status. */ + public DistributedMapStatus status() { + return summary.status(); + } + + /** Returns the completion reason. */ + public DistributedMapCompletionReason completionReason() { + return summary.completionReason(); + } + + /** Returns the count of succeeded items. */ + public long successCount() { + return summary.successCount(); + } + + /** Returns the count of failed items. */ + public long failureCount() { + return summary.failureCount(); + } + + /** Returns the count of unprocessed items. */ + public long unprocessedCount() { + return summary.unprocessedCount(); + } + + /** Returns the total item count, or null until known. */ + public Long totalCount() { + return summary.totalCount(); + } + + /** Returns the run ARN, or null when the run never started. */ + public String distributedMapRunArn() { + return summary.distributedMapRunArn(); + } + + /** Returns the run id derived from the ARN, or null when the run never started. */ + public String distributedMapId() { + return summary.distributedMapId(); + } + + /** Returns the completion details, or null when absent. */ + public String completionDetails() { + return summary.completionDetails(); + } + + /** Returns true when any item failed. */ + public boolean hasFailure() { + return summary.hasFailure(); + } + + /** Throws DistributedMapError when the run did not fully succeed, surfacing the first failed item. */ + public void throwIfError() { + if (summary.status() != DistributedMapStatus.SUCCEEDED) { + summary.throwIfError(); + return; + } + var failedItems = failed(); + if (!failedItems.isEmpty()) { + var first = failedItems.get(0); + throw DistributedMapError.itemLevel( + summary.status(), + summary.completionReason(), + summary.failureCount(), + first.error(), + first.itemId()); + } + if (summary.hasFailure()) { + summary.throwIfError(); + } + } +} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/model/DistributedMapResultItem.java b/sdk/src/main/java/software/amazon/lambda/durable/model/DistributedMapResultItem.java new file mode 100644 index 000000000..864cae769 --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/model/DistributedMapResultItem.java @@ -0,0 +1,23 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.model; + +/** Outcome of a single distributed map item. */ +public record DistributedMapResultItem(String itemId, Status status, O output, DistributedMapItemError error) { + + /** Status of an individual distributed map item. */ + public enum Status { + SUCCEEDED, + FAILED + } + + /** Creates a succeeded item. */ + public static DistributedMapResultItem succeeded(String itemId, O output) { + return new DistributedMapResultItem<>(itemId, Status.SUCCEEDED, output, null); + } + + /** Creates a failed item. */ + public static DistributedMapResultItem failed(String itemId, DistributedMapItemError error) { + return new DistributedMapResultItem<>(itemId, Status.FAILED, null, error); + } +} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/model/DistributedMapStatus.java b/sdk/src/main/java/software/amazon/lambda/durable/model/DistributedMapStatus.java new file mode 100644 index 000000000..f704ca353 --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/model/DistributedMapStatus.java @@ -0,0 +1,37 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.model; + +/** Terminal status of a distributed map run. */ +public enum DistributedMapStatus { + SUCCEEDED("SUCCEEDED"), + FAILED("FAILED"), + STOPPED("STOPPED"), + TIMED_OUT("TIMED_OUT"); + + private final String value; + + DistributedMapStatus(String value) { + this.value = value; + } + + /** Returns the wire-format string value. */ + public String getValue() { + return value; + } + + /** Returns the status matching a wire-format string value. */ + public static DistributedMapStatus fromValue(String value) { + for (var status : values()) { + if (status.value.equals(value)) { + return status; + } + } + throw new IllegalArgumentException("Unknown DistributedMapStatus: " + value); + } + + @Override + public String toString() { + return value; + } +} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/model/DistributedMapSummary.java b/sdk/src/main/java/software/amazon/lambda/durable/model/DistributedMapSummary.java new file mode 100644 index 000000000..8f26f03f0 --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/model/DistributedMapSummary.java @@ -0,0 +1,36 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.model; + +import software.amazon.lambda.durable.exception.DistributedMapError; + +/** Outcome of a distributed map run, without per-item results. */ +public record DistributedMapSummary( + DistributedMapStatus status, + DistributedMapCompletionReason completionReason, + long successCount, + long failureCount, + long unprocessedCount, + String distributedMapRunArn, + String completionDetails, + Long totalCount) { + + /** Returns the run id derived from the ARN, or null when the run never started. */ + public String distributedMapId() { + return distributedMapRunArn != null + ? distributedMapRunArn.substring(distributedMapRunArn.lastIndexOf(':') + 1) + : null; + } + + /** Returns true when any item failed. */ + public boolean hasFailure() { + return failureCount > 0; + } + + /** Throws DistributedMapError when the run did not fully succeed. */ + public void throwIfError() { + if (status != DistributedMapStatus.SUCCEEDED || hasFailure()) { + throw DistributedMapError.runLevel(status, completionReason, failureCount, completionDetails); + } + } +} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/model/OperationSubType.java b/sdk/src/main/java/software/amazon/lambda/durable/model/OperationSubType.java index c7273b9cd..2228b28f4 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/model/OperationSubType.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/model/OperationSubType.java @@ -22,7 +22,9 @@ public enum OperationSubType { PARALLEL_BRANCH(OperationType.CONTEXT, "ParallelBranch"), WAIT_FOR_CALLBACK(OperationType.CONTEXT, "WaitForCallback"), WAIT_FOR_CONDITION(OperationType.STEP, "WaitForCondition"), - WITH_RETRY(OperationType.CONTEXT, "WithRetry"); + WITH_RETRY(OperationType.CONTEXT, "WithRetry"), + // Requires OperationType.DISTRIBUTED_MAP, generated once the Lambda client ships the distributed-map shapes. + DISTRIBUTED_MAP(OperationType.DISTRIBUTED_MAP, "DistributedMap"); private final OperationType operationType; private final String value; diff --git a/sdk/src/main/java/software/amazon/lambda/durable/model/ReaderPage.java b/sdk/src/main/java/software/amazon/lambda/durable/model/ReaderPage.java new file mode 100644 index 000000000..173bac2d5 --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/model/ReaderPage.java @@ -0,0 +1,15 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.model; + +import java.util.Collections; +import java.util.List; + +/** A page of items and the next state returned by a distributed map reader function. */ +public record ReaderPage(List items, S nextState) { + + /** Applies a defensive copy and defaults. */ + public ReaderPage { + items = items != null ? List.copyOf(items) : Collections.emptyList(); + } +} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/operation/DistributedMapOperation.java b/sdk/src/main/java/software/amazon/lambda/durable/operation/DistributedMapOperation.java new file mode 100644 index 000000000..e912ef943 --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/operation/DistributedMapOperation.java @@ -0,0 +1,79 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.operation; + +// Written against DistributedMap* shapes not yet in the generated Lambda client, so this does not compile until they ship. + +import java.util.function.Function; +import software.amazon.awssdk.services.lambda.model.DistributedMapDetails; +import software.amazon.awssdk.services.lambda.model.DistributedMapOptions; +import software.amazon.awssdk.services.lambda.model.Operation; +import software.amazon.awssdk.services.lambda.model.OperationAction; +import software.amazon.awssdk.services.lambda.model.OperationUpdate; +import software.amazon.lambda.durable.TypeToken; +import software.amazon.lambda.durable.context.DurableContextImpl; +import software.amazon.lambda.durable.exception.DistributedMapException; +import software.amazon.lambda.durable.model.OperationIdentifier; +import software.amazon.lambda.durable.serde.SerDes; + +/** + * Durable operation that starts a distributed map run and waits for its terminal outcome. + * + * @param the resolved result type (a summary, or a result with collected items) + */ +public class DistributedMapOperation extends SerializableDurableOperation { + private final DistributedMapOptions options; + private final Function resultBuilder; + + public DistributedMapOperation( + OperationIdentifier operationIdentifier, + DistributedMapOptions options, + TypeToken resultType, + SerDes serDes, + Function resultBuilder, + DurableContextImpl durableContext) { + super(operationIdentifier, resultType, serDes, durableContext); + this.options = options; + this.resultBuilder = resultBuilder; + } + + /** Starts the operation. */ + @Override + protected void start() { + sendOperationUpdate(OperationUpdate.builder().action(OperationAction.START).distributedMapOptions(options)); + pollForOperationUpdates(); + } + + /** Replays the operation. */ + @Override + protected void replay(Operation existing) { + switch (existing.status()) { + case STARTED -> pollForOperationUpdates(); + case SUCCEEDED, FAILED, TIMED_OUT, STOPPED -> markAlreadyCompleted(); + default -> + throw terminateExecutionWithIllegalDurableOperationException( + "Unexpected distributed map status: " + existing.statusAsString()); + } + } + + /** Blocks until the operation completes and returns the resolved outcome. */ + @Override + public R get() { + var op = waitForOperationCompletion(); + return switch (op.status()) { + case SUCCEEDED -> { + var details = op.distributedMapDetails(); + if (details == null) { + throw terminateExecutionWithIllegalDurableOperationException( + "DISTRIBUTED_MAP operation succeeded but carried no DistributedMapDetails"); + } + yield resultBuilder.apply(details); + } + // Reaching here means the operation itself terminal-failed, so raise instead of hanging. + case FAILED, TIMED_OUT, STOPPED -> throw new DistributedMapException(op); + default -> + throw terminateExecutionWithIllegalDurableOperationException( + "Unexpected distributed map operation status: " + op.statusAsString()); + }; + } +} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/operation/DistributedMapWire.java b/sdk/src/main/java/software/amazon/lambda/durable/operation/DistributedMapWire.java new file mode 100644 index 000000000..efcd1ff9d --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/operation/DistributedMapWire.java @@ -0,0 +1,318 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.operation; + +// Written against DistributedMap* shapes not yet in the generated Lambda client, so this does not compile until they ship. + +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.List; +import software.amazon.awssdk.services.lambda.model.DistributedMapCsvDelimiter; +import software.amazon.awssdk.services.lambda.model.DistributedMapCsvFormatOptions; +import software.amazon.awssdk.services.lambda.model.DistributedMapCsvHeaderLocation; +import software.amazon.awssdk.services.lambda.model.DistributedMapDestinationInclude; +import software.amazon.awssdk.services.lambda.model.DistributedMapDestinationType; +import software.amazon.awssdk.services.lambda.model.DistributedMapDetails; +import software.amazon.awssdk.services.lambda.model.DistributedMapFunctionResponseType; +import software.amazon.awssdk.services.lambda.model.DistributedMapInlineSourceConfig; +import software.amazon.awssdk.services.lambda.model.DistributedMapOnFailureConfig; +import software.amazon.awssdk.services.lambda.model.DistributedMapOnSuccessConfig; +import software.amazon.awssdk.services.lambda.model.DistributedMapOptions; +import software.amazon.awssdk.services.lambda.model.DistributedMapProcessorConfig; +import software.amazon.awssdk.services.lambda.model.DistributedMapReaderFunctionSourceConfig; +import software.amazon.awssdk.services.lambda.model.DistributedMapResultCollectionConfig; +import software.amazon.awssdk.services.lambda.model.DistributedMapResultCollectionMode; +import software.amazon.awssdk.services.lambda.model.DistributedMapS3DestinationConfig; +import software.amazon.awssdk.services.lambda.model.DistributedMapS3SourceConfig; +import software.amazon.awssdk.services.lambda.model.DistributedMapS3SourceTransform; +import software.amazon.awssdk.services.lambda.model.DistributedMapSourceConfig; +import software.amazon.awssdk.services.lambda.model.DistributedMapSourceFormat; +import software.amazon.awssdk.services.lambda.model.DistributedMapSourceType; +import software.amazon.lambda.durable.TypeToken; +import software.amazon.lambda.durable.config.CsvFormat; +import software.amazon.lambda.durable.config.DistributedMapConfig; +import software.amazon.lambda.durable.config.DistributedMapProcessor; +import software.amazon.lambda.durable.config.DistributedMapSource; +import software.amazon.lambda.durable.config.FailureDestination; +import software.amazon.lambda.durable.config.ProcessorRetryConfig; +import software.amazon.lambda.durable.config.SuccessDestination; +import software.amazon.lambda.durable.exception.DurableExecutionException; +import software.amazon.lambda.durable.model.DistributedMapCompletionReason; +import software.amazon.lambda.durable.model.DistributedMapItemError; +import software.amazon.lambda.durable.model.DistributedMapResultItem; +import software.amazon.lambda.durable.model.DistributedMapStatus; +import software.amazon.lambda.durable.model.DistributedMapSummary; +import software.amazon.lambda.durable.model.DistributedMapResult; +import software.amazon.lambda.durable.serde.JacksonSerDes; +import software.amazon.lambda.durable.serde.SerDes; + +/** Translates distributed map config to the checkpoint options and the operation details to result types. */ +public final class DistributedMapWire { + private static final SerDes JSON = new JacksonSerDes(); + private static final long INLINE_SIZE_LIMIT = 1024L * 1024L; + private static final long READER_STATE_LIMIT = 32L * 1024L; + + private DistributedMapWire() {} + + public static DistributedMapOptions toOptions( + DistributedMapSource source, + DistributedMapProcessor processor, + int maxConcurrency, + DistributedMapConfig config, + SerDes defaultSerDes, + boolean collectResults) { + var builder = DistributedMapOptions.builder() + .maxConcurrency(maxConcurrency) + .source(sourceConfig(source, defaultSerDes)) + .processor(processorConfig(processor)); + if (config.destination() != null) { + var destination = destinationConfig(config.destination()); + if (destination != null) { + builder.destination(destination); + } + } + var completion = completionConfig(config); + if (completion != null) { + builder.completionConfig(completion); + } + if (collectResults) { + builder.resultCollection(DistributedMapResultCollectionConfig.builder() + .mode(DistributedMapResultCollectionMode.INLINE) + .build()); + } + if (config.timeout() != null) { + builder.timeoutSeconds((int) config.timeout().toSeconds()); + } + return builder.build(); + } + + private static DistributedMapSourceConfig sourceConfig(DistributedMapSource source, SerDes defaultSerDes) { + var builder = DistributedMapSourceConfig.builder() + .type(DistributedMapSourceType.fromValue(source.sourceType().name())); + if (source.maxItems() != null) { + builder.maxItemsToRead(source.maxItems().longValue()); + } + switch (source.sourceType()) { + case INLINE -> builder.inlineSourceConfig(inlineSourceConfig(source, defaultSerDes)); + case S3 -> builder.s3SourceConfig(s3SourceConfig(source.s3())); + case READER_FUNCTION -> builder.readerFunctionSourceConfig(readerConfig(source.reader(), defaultSerDes)); + } + return builder.build(); + } + + private static DistributedMapInlineSourceConfig inlineSourceConfig( + DistributedMapSource source, SerDes defaultSerDes) { + var serdes = source.inlineSerDes() != null ? source.inlineSerDes() : defaultSerDes; + var bodies = new ArrayList(source.inlineItems().size()); + for (Object item : source.inlineItems()) { + bodies.add(serdes.serialize(item)); + } + var arrayBytes = JSON.serialize(bodies).getBytes(StandardCharsets.UTF_8).length; + if (arrayBytes > INLINE_SIZE_LIMIT) { + throw new IllegalArgumentException( + "inline source exceeds the 1 MB limit (serialized size: " + arrayBytes + " bytes)"); + } + return DistributedMapInlineSourceConfig.builder().items(bodies).build(); + } + + private static DistributedMapS3SourceConfig s3SourceConfig(DistributedMapSource.S3SourceConfig s3) { + var builder = DistributedMapS3SourceConfig.builder().bucket(s3.bucket()); + if (s3.key() != null) { + builder.key(s3.key()); + } + if (s3.prefix() != null) { + builder.keyPrefix(s3.prefix()); + } + if (s3.transform() != null) { + builder.transform(DistributedMapS3SourceTransform.fromValue(s3.transform().name())); + } + if (s3.format() != null) { + builder.format(DistributedMapSourceFormat.fromValue(s3.format().name())); + } + if (s3.expectedBucketOwner() != null) { + builder.expectedBucketOwner(s3.expectedBucketOwner()); + } + if (s3.csvFormat() != null) { + builder.csvFormatOptions(csvFormatOptions(s3.csvFormat())); + } + return builder.build(); + } + + private static DistributedMapCsvFormatOptions csvFormatOptions(CsvFormat format) { + var builder = DistributedMapCsvFormatOptions.builder() + .headerLocation(DistributedMapCsvHeaderLocation.fromValue(format.headerLocation().name())) + .delimiter(DistributedMapCsvDelimiter.fromValue(format.delimiter().getValue())); + // Column names go on the wire only for headerless files (GIVEN), expectedColumns (FIRST_ROW) is compile-time only. + if (format.headerLocation() == CsvFormat.HeaderLocation.GIVEN && format.columns() != null) { + builder.headers(format.columns()); + } + return builder.build(); + } + + private static DistributedMapReaderFunctionSourceConfig readerConfig( + DistributedMapSource.ReaderSourceConfig reader, SerDes defaultSerDes) { + var builder = DistributedMapReaderFunctionSourceConfig.builder().functionName(reader.functionName()); + if (reader.initialState() != null) { + var serdes = reader.stateSerDes() != null ? reader.stateSerDes() : defaultSerDes; + var state = serdes.serialize(reader.initialState()); + if (state != null && state.getBytes(StandardCharsets.UTF_8).length > READER_STATE_LIMIT) { + throw new IllegalArgumentException("reader initialState exceeds the 32 KB limit"); + } + builder.initialState(state); + } + return builder.build(); + } + + private static DistributedMapProcessorConfig processorConfig(DistributedMapProcessor processor) { + var builder = DistributedMapProcessorConfig.builder().functionName(processor.functionName()); + if (processor.durableExecutionNamePrefix() != null) { + builder.durableExecutionNamePrefix(processor.durableExecutionNamePrefix()); + } + var responseType = processor.responseMode().getValue(); + if (responseType != null) { + builder.functionResponseTypes(List.of(DistributedMapFunctionResponseType.fromValue(responseType))); + } + if (processor.batchSize() != null) { + builder.batchSize(processor.batchSize()); + } + ProcessorRetryConfig retry = processor.retryConfig(); + if (retry != null) { + if (retry.maxRetryAttempts() != null) { + builder.maxRetryAttempts(retry.maxRetryAttempts()); + } + if (retry.maxRetryDuration() != null) { + builder.maxRetryDurationSeconds((int) retry.maxRetryDuration().toSeconds()); + } + } + return builder.build(); + } + + private static software.amazon.awssdk.services.lambda.model.DistributedMapDestinationConfig destinationConfig( + software.amazon.lambda.durable.config.DistributedMapDestinationConfig destination) { + if (destination.onSuccess() == null && destination.onFailure() == null) { + return null; // omit an empty destination config from the wire + } + var builder = software.amazon.awssdk.services.lambda.model.DistributedMapDestinationConfig.builder(); + SuccessDestination onSuccess = destination.onSuccess(); + if (onSuccess != null) { + var include = new ArrayList(); + if (onSuccess.includeInput()) { + include.add(DistributedMapDestinationInclude.INPUT); + } + if (onSuccess.includeOutput()) { + include.add(DistributedMapDestinationInclude.OUTPUT); + } + builder.onSuccess(DistributedMapOnSuccessConfig.builder() + .type(DistributedMapDestinationType.S3) + .include(include) + .s3DestinationConfig(s3Destination( + onSuccess.bucket(), onSuccess.prefix(), onSuccess.expectedBucketOwner())) + .build()); + } + FailureDestination onFailure = destination.onFailure(); + if (onFailure != null) { + var include = new ArrayList(); + if (onFailure.includeInput()) { + include.add(DistributedMapDestinationInclude.INPUT); + } + if (onFailure.includeError()) { + include.add(DistributedMapDestinationInclude.ERROR); + } + builder.onFailure(DistributedMapOnFailureConfig.builder() + .type(DistributedMapDestinationType.S3) + .include(include) + .s3DestinationConfig(s3Destination( + onFailure.bucket(), onFailure.prefix(), onFailure.expectedBucketOwner())) + .build()); + } + return builder.build(); + } + + private static DistributedMapS3DestinationConfig s3Destination( + String bucket, String keyPrefix, String expectedBucketOwner) { + var builder = DistributedMapS3DestinationConfig.builder().bucket(bucket).keyPrefix(keyPrefix); + if (expectedBucketOwner != null) { + builder.expectedBucketOwner(expectedBucketOwner); + } + return builder.build(); + } + + private static software.amazon.awssdk.services.lambda.model.DistributedMapCompletionConfig completionConfig( + DistributedMapConfig config) { + var completion = config.completionConfig(); + if (completion == null + || (completion.toleratedFailureCount() == null + && completion.toleratedFailurePercentage() == null + && completion.minimumSampleSize() == null)) { + return null; // omit an empty completion config from the wire + } + var builder = software.amazon.awssdk.services.lambda.model.DistributedMapCompletionConfig.builder(); + if (completion.toleratedFailureCount() != null) { + builder.toleratedFailureCount(completion.toleratedFailureCount()); + } + if (completion.toleratedFailurePercentage() != null) { + builder.toleratedFailurePercentage(completion.toleratedFailurePercentage().floatValue()); + } + if (completion.minimumSampleSize() != null) { + builder.minimumSampleSize(completion.minimumSampleSize()); + } + return builder.build(); + } + + public static DistributedMapSummary toSummary(DistributedMapDetails details) { + return new DistributedMapSummary( + parseStatus(details), + parseReason(details), + details.successCount(), + details.failureCount(), + details.unprocessedCount(), + details.distributedMapRunArn(), + details.completionDetails(), + details.totalCount()); + } + + public static DistributedMapResult toResult( + DistributedMapDetails details, TypeToken resultType, SerDes serdes) { + var summary = toSummary(details); + var items = new ArrayList>(); + if (details.results() != null) { + for (var wire : details.results()) { + var status = wire.statusAsString(); + if (DistributedMapResultItem.Status.SUCCEEDED.name().equals(status)) { + O output = wire.output() != null ? serdes.deserialize(wire.output(), resultType) : null; + items.add(DistributedMapResultItem.succeeded(wire.itemId(), output)); + } else if (DistributedMapResultItem.Status.FAILED.name().equals(status)) { + var error = wire.error() != null + ? new DistributedMapItemError( + wire.error().errorType() != null ? wire.error().errorType() : "", + wire.error().errorMessage() != null ? wire.error().errorMessage() : "") + : null; + items.add(DistributedMapResultItem.failed(wire.itemId(), error)); + } else { + throw new DurableExecutionException("unrecognized distributed map item status from the backend (" + + status + ") for item " + wire.itemId()); + } + } + } + return new DistributedMapResult<>(summary, items); + } + + private static DistributedMapStatus parseStatus(DistributedMapDetails details) { + if (details.statusAsString() == null) { + throw new DurableExecutionException("distributed map details missing the required Status field"); + } + try { + return DistributedMapStatus.fromValue(details.statusAsString()); + } catch (IllegalArgumentException e) { + throw new DurableExecutionException( + "unrecognized distributed map status from the backend (" + details.statusAsString() + ")", e); + } + } + + private static DistributedMapCompletionReason parseReason(DistributedMapDetails details) { + if (details.completionReasonAsString() == null) { + throw new DurableExecutionException("distributed map details missing the required CompletionReason field"); + } + return DistributedMapCompletionReason.fromValue(details.completionReasonAsString()); + } +} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/util/DistributedMapValidation.java b/sdk/src/main/java/software/amazon/lambda/durable/util/DistributedMapValidation.java new file mode 100644 index 000000000..f5dbba112 --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/util/DistributedMapValidation.java @@ -0,0 +1,62 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.util; + +import java.util.HashSet; +import java.util.List; +import java.util.regex.Pattern; + +/** Shared validation and S3 URI parsing for distributed map configuration. */ +public final class DistributedMapValidation { + private static final Pattern ACCOUNT_ID = Pattern.compile("\\d{12}"); + private static final String S3_SCHEME = "s3://"; + private static final int MAX_FUNCTION_NAME_LENGTH = 170; + + private DistributedMapValidation() {} + + public static void validateFunctionName(String functionName) { + if (functionName == null || functionName.isBlank()) { + throw new IllegalArgumentException("function name cannot be empty"); + } + if (functionName.length() > MAX_FUNCTION_NAME_LENGTH) { + throw new IllegalArgumentException( + "function name must be at most 170 characters, got: " + functionName.length()); + } + } + + public static void validateBucketOwner(String expectedBucketOwner) { + if (expectedBucketOwner != null + && !ACCOUNT_ID.matcher(expectedBucketOwner).matches()) { + throw new IllegalArgumentException( + "expectedBucketOwner must be a 12-digit account id, got: " + expectedBucketOwner); + } + } + + public static void validateColumns(String parameterName, List columns) { + if (columns == null) { + return; + } + if (columns.isEmpty()) { + throw new IllegalArgumentException(parameterName + " cannot be empty"); + } + if (new HashSet<>(columns).size() != columns.size()) { + throw new IllegalArgumentException(parameterName + " must not contain duplicates"); + } + } + + public static ParsedS3Uri parseS3Uri(String uri) { + if (uri == null || !uri.startsWith(S3_SCHEME)) { + throw new IllegalArgumentException("S3 URI must start with s3://, got: " + uri); + } + var remainder = uri.substring(S3_SCHEME.length()); + var slash = remainder.indexOf('/'); + var bucket = slash < 0 ? remainder : remainder.substring(0, slash); + var path = slash < 0 ? null : remainder.substring(slash + 1); + if (bucket.isEmpty()) { + throw new IllegalArgumentException("S3 URI must contain a bucket, got: " + uri); + } + return new ParsedS3Uri(bucket, path == null || path.isEmpty() ? null : path); + } + + public record ParsedS3Uri(String bucket, String path) {} +}