feat(sdk): add distributed map operation - #1
Conversation
| 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 { |
There was a problem hiding this comment.
This isn't an operation. Why is this class in operation package ?
| 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 { |
There was a problem hiding this comment.
So this is a helper that help users to create processor or reader easier. Users can create their own without this? If so, why does it belong to the root package?
| ITEMS_OP_NAME, | ||
| bodies, | ||
| OBJECT_TYPE, | ||
| (body, index, mapContext) -> |
There was a problem hiding this comment.
[P1] Normalize results before the durable map checkpoints them. resultSerDes is only used later when constructing the processor response. This ctx.map has no SerDes configured, so DurableContextImpl checkpoints each O with the helper's default Jackson SerDes and replays it as a generic Object. A custom SerDes can therefore fail before it is called or produce different JSON after a suspension—for example, POJO naming rules are lost once the value is a Map. Convert each result to its wire JSON value inside the map item before it is checkpointed, or otherwise checkpoint it with the supplied SerDes and a stable result type.
| 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]))); |
There was a problem hiding this comment.
[P2] Keep output serialization inside the per-item failure boundary. Input conversion and func.apply failures are captured through each Future, but toJsonValue runs afterward. If one successful function returns a value that resultSerDes cannot serialize, this throws from the handler and causes the entire batch to fail/retry instead of adding only that item to batchItemFailures. Convert and store the wire output inside the submitted task so serialization errors are reported for the corresponding item.
| var outSerdes = resultSerDes != null ? resultSerDes : JSON; | ||
| return (event, context) -> { | ||
| var records = records(event); | ||
| var workers = concurrency > 0 ? concurrency : Math.max(1, records.size()); |
There was a problem hiding this comment.
[P2] Reject or bound nonpositive concurrency. A zero or negative value falls back to one platform thread per record. Since this PR permits processor batches of up to 10,000 items, a misconfigured handler can attempt to create 10,000 threads and fail with native-thread or memory exhaustion. Require concurrency >= 1 or choose a bounded default such as the available processor count.
|
|
||
| /** Returns the outputs of succeeded items. */ | ||
| public List<O> getResults() { | ||
| return succeeded().stream().map(DistributedMapResultItem::output).filter(Objects::nonNull).toList(); |
There was a problem hiding this comment.
[P2] Preserve successful null outputs. Filtering null removes legitimate successful results and changes the list's cardinality, so callers can no longer align outputs with succeeded() items. The existing MapResult.succeeded() API explicitly preserves null results; this convenience method should do the same or clearly expose a differently named non-null-only view.
| */ | ||
| public class DistributedMapException extends DurableOperationException { | ||
| public DistributedMapException(Operation operation) { | ||
| super( |
There was a problem hiding this comment.
[P2] Preserve an error object for the outer execution failure. DurableExecutor.buildErrorObject special-cases every DurableOperationException by returning getErrorObject(). Passing null here therefore turns an uncaught distributed-map failure into DurableExecutionOutput.failure(null), discarding this message from the backend response. Synthesize an ErrorObject from the status/message, use an error carried by the operation details, or avoid the special-cased exception hierarchy.
| * 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 { |
There was a problem hiding this comment.
Name is confusing: DistributedMapError and DistributedMapException. Maybe DistributedMapUserException instead of DistributedMapError?
| } | ||
|
|
||
| /** Wraps a per-item function as a processor handler. Use REPORT_ITEM_RESULTS or REPORT_FAILED_ITEMS. */ | ||
| public static <I, O> RequestHandler<Map<String, Object>, Map<String, Object>> createDistributedMapItemHandler( |
There was a problem hiding this comment.
Can this be concrete types instead of Object?
2 participants
By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
Issue Link, if available
N/A
Description
Adds the distributed map operation (
ctx.distributedMap) to the Java SDK.A map run processes a bounded dataset in parallel. A customer starts a map run from a durable function, naming a source to read items from, a processor function to invoke per batch, and concurrency, retry, and failure settings. The service reads items from the source, groups them into batches, invokes the processor for each batch, retries failures, tracks progress, routes successful results and failed items to destinations, and reports completion.
Changes:
model/: result types (DistributedMapSummary,DistributedMapResult,DistributedMapResultItem,DistributedMapItemError) and the run enums (DistributedMapStatus,DistributedMapCompletionReason).config/:DistributedMapConfig,DistributedMapSource,DistributedMapProcessor,ProcessorRetryConfig,DistributedMapCompletionConfig, the destination types (SuccessDestination,FailureDestination,DistributedMapDestinationConfig,DistributedMapDestination), and CSV options (CsvFormat,CsvDelimiter).DurableContext.java/context/DurableContextImpl.java: thectx.distributedMapentry point, with overloads for summary vs result-collecting,Class<O>vsTypeToken<O>, blocking vs async, and optional config.DistributedMapHandlers.java(andmodel/ReaderPage.java): authoring wrappers so a customer can write a plain function and use it as a processor or reader Lambda without hand-writing the item or batch protocol, including durable-execution variants.operation/DistributedMapOperation.java: drives the run so the caller suspends while it executes and resumes with the finished outcome, raising a clear error if the operation itself fails.operation/DistributedMapWire.java: maps config ontoDistributedMapOptionsand parsesDistributedMapDetailsback into the result types.exception/:DistributedMapError(run or item failure) andDistributedMapException(the operation itself terminal-failing).util/DistributedMapValidation.java: shared validation and S3 URI parsing.WIP STATUS: this PR cannot compile yet. We are still waiting for the distributed map wire shapes to be released to the AWS SDK for the Java Lambda client. This code is written against those shapes, but to compile we need to bump the SDK's minimum
software.amazon.awssdk:lambdaversion once the new models are released.I am looking into building against a local Lambda client updated with the distributed map shapes, so we can run tests ahead of the official AWS SDK release.
Demo/Screenshots
N/A
Checklist
Testing
Unit Tests
Have unit tests been written for these changes?
TODO
Integration Tests
Have integration tests been written for these changes?
TODO
Examples
Has a new example been added for the change? (if applicable)
TODO