feat(sdk): add distributed map operation by nvasiu · Pull Request #1 · nvasiu/aws-durable-execution-sdk-java

nvasiu · 2026-08-28T23:20:04Z

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/

The result types a customer receives back from a map run.

DistributedMapSummary: what ctx.distributedMap returns, describes the run's overall outcome. Carries the status, the item counts, sourceExhausted, and a run-level error. throwIfError() opts into raising.
DistributedMapResult: returned when a result type is passed. Contains individual map run item outcomes.
DistributedMapResult.Item: represents a single item's result. Nested, since it belongs to exactly one result.
DistributedMapError: the error record, with errorType, errorMessage, errorData and stackTrace. Top-level, since both the summary and each item carry one.
DistributedMapStatus: the run enum.

config/

The input types a customer constructs to describe a map run.

DistributedMapConfig: optional settings for distributed map, nesting CompletionConfig for the item failure thresholds that mark the overall run failed. toleratedFailurePercentage is an integer from 0 through 100.
DistributedMapSource: describes where map run items come from, with factories for inline, S3 and reader sources, and nesting the CSV options. An S3 source carries a source type with one matching nested config block, so a listed-object option cannot be set on a single-object source.
DistributedMapProcessor: describes the Lambda that processes items, and how outcomes are reported back via reportBatchOutcome and reportItemOutcomes.
DistributedMapProcessorRetryConfig: the processor's per-item retry policy, holding maxRetryAttempts and maxRetryDuration. Retry belongs to the processor reference rather than the run, because it defines how the service invokes that function, so one Lambda can back several references with different policies. UNLIMITED means no attempt limit within the duration budget.
DistributedMapDestination: routes successful and failed item records to S3 through s3Successes and s3Failures, nesting Success and Failure so a success destination cannot be used for failures.

DurableContext.java and context/DurableContextImpl.java

The customer-facing entry point on the durable execution context.

ctx.distributedMap: the method a customer calls to run a distributed map. Overloaded so the return type is DistributedMapResult when a result type is passed and DistributedMapSummary otherwise, with Class<O> and TypeToken<O> forms, blocking and async forms, and optional config.
Validates that maxConcurrency is greater than zero. The service enforces the upper bound.

dmap/

Authoring helpers for the processor and reader Lambdas, so a customer writes a plain function rather than the item or batch protocol.

DistributedMapHandlers: createDistributedMapItemHandler, createDistributedMapBatchHandler, createDistributedMapReader, and the durable variants createDistributedMapItemHandlerWithDurableExecution and createDistributedMapBatchHandlerWithDurableExecution.
The item handlers run at most concurrency handlers at a time, defaulting to one.
Both durable variants accept a DurableConfig, so a durable processor can register plugins and supply its own client.
An item handler that returns nothing reports no output for that item, rather than passing the null to the result serdes.
A reported item failure carries the error's stack trace.
ProcessorEvent, ProcessorRecord, ItemHandlerResponse, ItemResult, ItemFailure, ReaderEvent, ReaderResponse: the wire envelopes as records, so the handler signatures name concrete types and the Lambda runtime handles serialization. Envelope validation lives in their compact constructors, which the runtime invokes when it deserializes.
ReaderPage: what a customer's reader function returns, items plus the next state.

operation/DistributedMapOperation.java

The executor that drives the operation against the durable execution runtime.

Suspends the caller's function while the run executes and resumes it with the finished outcome.
Every terminal state resolves, and throwIfError() opts into raising.
A replay that finds the operation PENDING waits for the terminal outcome. The start checkpoint records the operation as PENDING, and it reaches STARTED only once the service confirms the run was created, so a replay can land in that window.
A terminal operation missing SuccessCount, FailureCount or UnprocessedCount throws, naming the missing count. A SUCCEEDED run missing TotalCount throws too.
All translation between the config types and the service shapes happens here, mapping onto DistributedMapOptions and parsing DistributedMapDetails back.
Items, Output and the record body are opaque strings, matching the API model, so any serdes works.

operation/ChildContextOperation.java

Operation type handling.

Handles DISTRIBUTED_MAP in the operation sub-type switch.

exception/DistributedMapException.java

The error type a customer catches.

Thrown when a run or an item fails, carrying the run status, the error object and the failure count.

util/DistributedMapValidation.java

Shared validation and S3 URI parsing.

maxItems is an integer from 1 through 100,000,000.
A CSV column list holds at most 1,000 unique names of at most 256 characters.
A destination key prefix is at most 900 UTF-8 bytes, counting any trailing slash. Bytes are counted rather than characters, because a multi-byte prefix under 900 characters can still exceed 900 bytes.

execution/CheckpointManager.java

Batching of operation updates into checkpoint requests.

estimateSize() now counts a distributed map START update's inline source items, so a large inline source is no longer measured as about 100 bytes when updates are packed into a checkpoint request.

Tests

sdk/src/test/.../config/

Config and argument validation: DistributedMapConfigTest (11), DistributedMapSourceTest (40), DistributedMapProcessorTest (16), DistributedMapProcessorRetryConfigTest (7), DistributedMapDestinationTest (19).

sdk/src/test/.../model/

The result types and the status enum: DistributedMapResultTest (11), DistributedMapSummaryTest (7), DistributedMapStatusTest (4).

sdk/src/test/.../operation/

Executor behaviour and serialization round trips: DistributedMapOperationTest (11), DistributedMapOperationTranslationTest (54), which asserts the wire shape in both directions.

sdk/src/test/.../context/DurableContextDistributedMapTest.java (8)

The ctx.distributedMap surface and its validation.

sdk/src/test/.../dmap/DistributedMapHandlersTest.java (24) and DistributedMapDurableHandlersTest.java (2)

Authoring helper tests: checking that they process items, report failures against the right item, reject bad inputs, and emit the exact JSON the Lambda runtime sends. The concurrency tests use a barrier, so they fail if the pool does not genuinely run items in parallel. The durable tests invoke each durable wrapper end to end and assert the config reaches the executor.

NOTE: This branch does not compile from a clean checkout, since the distributed map shapes haven't been released to a public AWS SDK version. I build and test against a locally patched Lambda client that carries them. The pin needs bumping to the first released version with those shapes before this can merge.

Demo/Screenshots

N/A

Checklist

I have filled out every section of the PR template
I have thoroughly tested this change

Testing

Unit Tests

Have unit tests been written for these changes?

See above.

Integration Tests

Have integration tests been written for these changes?

TODO

Examples

Has a new example been added for the change? (if applicable)

TODO

Future Tasks

Add distributed map to the local emulator, then add full end to end tests against it.
Add distributed map examples to the examples package.

nvasiu · 2026-09-25T18:45:52Z

Refactored to address @zhongkechen comments. The following changes were made:

Fixes

Terminal states resolve with the summary, throwIfError() is the opt-in
Removed double serialization at three sites (inline items, toItem, output)
maxConcurrency validates 1 to 10000
Run status comes from the operation, parseStatus deleted
Durable item handler serializes before checkpointing, so a custom resultSerDes survives a suspension
Output serialization moved inside each task, so one bad result fails its own item
Non-positive concurrency rejected, new overload defaults to one
getResults() preserves null outputs

Deletions

DistributedMapWire, folded into DistributedMapOperation
ProcessorRetryConfig, folded onto DistributedMapProcessor
The original DistributedMapException, was never thrown
distributedMapId from the summary and result, removed because it's not useful
toOutputBody, four from(...) parsers, five toMap(...) builders

Renames

DistributedMapError to DistributedMapException
reportBatchOutcome / reportFailedItems / reportItemResults to batch / itemFailures / itemResults, wire strings unchanged

Nesting, eight top-level names removed

DistributedMapResultItem and DistributedMapItemError into DistributedMapResult
DistributedMapCompletionConfig into DistributedMapConfig
CsvFormat and CsvDelimiter into DistributedMapSource
SuccessDestination, FailureDestination and DistributedMapDestinationConfig collapsed into DistributedMapDestination

Package and types

DistributedMapHandlers from the root package, and ReaderPage from model/, into dmap/
Seven envelope records added in dmap/, handler signatures name them instead of Map and Object
ItemFailure nests ErrorDetail, ReaderEvent.maxItems boxed, ReaderResponse generic
Envelope validation in compact constructors
DISTRIBUTED_MAP added to the ChildContextOperation switch

Tests

Inverted two tests that asserted the old null filtering
Folded DistributedMapItemErrorTest and CsvFormatTest into their owners
DistributedMapCompletionConfigTest to DistributedMapConfigTest, DistributedMapWireTest to DistributedMapOperationTranslationTest
New: per-item serialization boundary, parameterized non-positive concurrency, peak-in-flight default, barrier-based parallelism, three wire-shape tests

zhongkechen · 2026-10-01T20:21:18Z

+                    var index = i;
+                    futures.add(pool.submit(() -> {
+                        I item = toItem(inSerdes, records.get(index).body(), itemType);
+                        outputs[index] = outSerdes.serialize(func.apply(item));


Is the output used for failed items?

No, outputs is only supposed to contain the results of succeeded items. If the serialization fails here, it will throw and the serialization error will be added to errors instead.

But there is a related issue here. In the ITEM_FAILURES mode, we don't need to return outputs but we still try to serialize each item. This was causing successful items to become failed.

Adding an update to skip this unused serialization when in ITEM_FAILURES mode.

zhongkechen · 2026-10-01T20:25:00Z

+        return (event, context) -> {
+            S state = event.state() != null ? serdes.deserialize(event.state(), stateType) : null;
+
+            var page = func.apply(state);


how would the reader know maxItems?

This is actually an oversight, thanks for pointing it out. The service is supposed to pass the max items per page when invoking the reader function. I will update this to accept the max items.

zhongkechen · 2026-10-01T20:28:04Z

+                    for (var record : records) {
+                        bodies.add(record.body());
+                    }
+                    MapResult<String> batch = ctx.map(


How many threads would be created at maximum for this map operation?

One thread per item + one thread for ctx.map itself. Concurrency is currently unbounded in this durable helper, so a thread could be started for every item in the batch (max size 10,000).

I will add a max concurrency option to this helper, and give it a default value of 1 (matching the non-durable version of this helper).

zhongkechen · 2026-10-01T20:29:59Z

+    @Override
+    protected void start() {
+        sendOperationUpdate(
+                OperationUpdate.builder().action(OperationAction.START).distributedMapOptions(options));


Is it possible to exceed the checkpoint size limit here?

Dmap options can include an inline list of items, with a max size of 1MB. This won't exceed the checkpoint size limit. But there is a related issue:

This SDK batches checkpoint updates together, adding updates while the batch stays under 750kb, and estimateSize() is what measures each update.

But estimateSize() doesn't look at an operation's options, because before dmap no customer-controlled data went into them. So an inline list of up to 1MB will be ignored when batching, and we can end up with batches much larger than we expect.

I will update estimateSize() to check dmap's inline items.

nvasiu · 2026-10-07T18:48:39Z

New revision to address recent comments:

dmap/DistributedMapHandlers.java: stopped serializing item outputs when the processor reports ITEM_FAILURES.
dmap/DistributedMapHandlers.java: passed the page bound to the reader function, which now takes BiFunction<S, Integer, ReaderPage<I, S>>.
dmap/ReaderEvent.java: renamed maxItems to maxItemsPerPage, matching the key the service sends.
dmap/DistributedMapHandlers.java: added a concurrency parameter to createDistributedMapItemHandlerWithDurableExecution, defaulting to 1 and validated by the same check the non-durable handler uses.
execution/CheckpointManager.java: estimateSize() counts a distributed map START update's inline source items.

Other changes:

config/DistributedMapSource.java: renamed maxItemsToRead to maxItems.
model/DistributedMapResult.java: renamed getResults to results and getErrors to errors.
model/DistributedMapResult.java: dropped the null filter from errors, so it returns one entry per failed item, including items that failed without error details.
exception/DistributedMapException.java: replaced the runLevel and itemLevel factories with two public constructors, matching how the sibling exceptions are built.
dmap/ItemFailure.java: trimmed the javadoc.
dmap/DistributedMapHandlers.java: added an itemSerDes parameter to createDistributedMapReader and serialized each item of a reader page with it, so the processor deserializes item bodies with the serdes that encoded them. ReaderResponse.items is now List<String>.
Tests updated for the renames, plus new coverage for the reader page bound, the durable handler's concurrency validation, the inline source's effect on batch packing, and the reader-to-processor round trip under a custom item serdes.

nvasiu · 2026-10-17T00:08:32Z

Another new revision to align with updated functional spec:

Note: These changes target wire shapes that aren't released to the Lambda client yet, so this branch builds only against a locally patched client and will not compile from a clean checkout until that client ships.

Model and API changes

config/DistributedMapProcessor.java: ResponseMode is BATCH or ITEM_OUTCOMES, sending the single wire value REPORT_BATCH_ITEM_OUTCOMES. The itemFailures and itemResults factories are replaced by itemOutcomes.
dmap/DistributedMapHandlers.java: the report parameter comes off both item handlers and validateItemReport is deleted, so both wrappers always return batchItemFailures and batchItemResults.
dmap/ItemHandlerResponse.java: both factories are deleted, and the wrappers construct the record directly.
config/DistributedMapSource.java: S3SourceType replaces Transform, and S3SourceConfig splits into ObjectConfig, ListedObjectMetadataConfig and ListedObjectsConfig. The seven public factory names are unchanged.
model/DistributedMapSummary.java: sourceExhausted and error replace completionReason and completionDetails.
model/DistributedMapCompletionReason.java: deleted, all 13 values.
model/DistributedMapError.java: new record with errorType, errorMessage, errorData and stackTrace, shared by the summary and each item. It replaces DistributedMapResult.ItemError.
model/DistributedMapResult.java: sourceExhausted() and error() replace completionReason() and completionDetails(), and errors() returns List<DistributedMapError>.
exception/DistributedMapException.java: carries the error object in place of a completion reason and a details string.
operation/DistributedMapOperation.java: s3SourceConfig sets the source type and the one matching block, the destination includes come from two enums, and sourceExhausted is required at a terminal state only when the run ARN is present.
dmap/ItemFailure.java: ErrorDetail gains errorData and stackTrace.
config/DistributedMapConfig.java: toleratedFailurePercentage is an integer.

Fixes

dmap/DistributedMapHandlers.java: an item handler that returns nothing reports no output, instead of passing the null to the result serdes.
dmap/ItemFailure.java: a reported item failure carries the error's stack trace.
operation/DistributedMapOperation.java: a replay that finds the operation PENDING waits for the terminal outcome, instead of terminating the execution.
operation/DistributedMapOperation.java: an absent terminal count reads as zero. Reading the boxed value into a primitive threw a NullPointerException.
operation/DistributedMapOperation.java: an item error carries all four members, and a null errorType or errorMessage stays null instead of becoming an empty string.

Validation

config/DistributedMapSource.java: maxItems is bounded at 1 through 100,000,000.
util/DistributedMapValidation.java: a CSV column list holds at most 1,000 names of at most 256 characters.
context/DurableContextImpl.java: the maxConcurrency upper bound is removed. The SDK checks only that the value is greater than zero, and the service enforces the range.

Documentation

DurableContext.java: maxConcurrency is documented as required with no service default, on all twelve overloads.
exception/DistributedMapException.java: the class javadoc covers the item-level constructor.
dmap/ItemResult.java: the output javadoc states that a null means the handler returned nothing and is omitted from the JSON.

Tests

Updated for the renames and the new shapes, plus new coverage for an unrecognized item status, a null item status, an item handler returning nothing, a captured stack trace, the maxItems and CSV column bounds, a PENDING replay, and absent terminal counts.

nvasiu · 2026-10-21T21:21:03Z

One more update to add some minor fixes from reviews for other SDKs:

Fixes

operation/DistributedMapOperation.java: a terminal operation missing SuccessCount, FailureCount or UnprocessedCount throws, naming the missing count. A SUCCEEDED run missing TotalCount throws too.
util/DistributedMapValidation.java: new validateDestinationPrefix rejects a destination key prefix over 900 UTF-8 bytes, counting bytes and any trailing slash.
config/DistributedMapDestination.java: the Success and Failure constructors call validateDestinationPrefix, so a too-long prefix is rejected when the destination is created rather than when the start checkpoint is written.
dmap/DistributedMapHandlers.java: both durable wrappers take a DurableConfig and pass it to DurableExecutor.wrap, so a durable processor can register plugins and supply its own client. Four durable item-handler overloads and two durable batch-handler overloads. A null config uses DurableConfig.defaultConfig().

Tests

One test per missing count, plus a SUCCEEDED run with no TotalCount and a FAILED run omitting it.
Prefix at the limit, over the limit on both factories, a two-byte-per-character prefix proving bytes are counted, and one that fails only with the trailing slash.
Each durable wrapper invoked end to end, asserting the config reaches the executor.

Addresses review feedback on the unreleased distributed map feature. Fixes: - The durable item handler checkpointed item results with the default serdes, so a custom resultSerDes produced different JSON after a suspension. Results are now serialized before being checkpointed. - Output serialization ran after the per-item futures settled, so one unserializable result failed the whole batch. It now runs inside each submitted task, so the failure is reported against its own item. - Non-positive concurrency fell back to one thread per record. It is now rejected, and a new overload defaults to one. - getResults filtered out null outputs, so the list no longer lined up with succeeded(). It now preserves them, matching MapResult. - An item handler that returns nothing reports no output. - A reported item failure carries the error's stack trace. - A replay that finds the operation PENDING waits for the outcome. - An absent terminal count reads as zero rather than throwing. - maxItems is bounded at 100,000,000, and a CSV column list at 1,000 names of at most 256 characters. - The maxConcurrency upper bound is removed, leaving the service to enforce the range. Changes: - Authoring helpers move into a dmap package, keeping feature specific classes off the root package. - Handler signatures use concrete records instead of Map and Object. The seven envelope types are public, and envelope validation moves into their compact constructors. - DistributedMapWire folds into DistributedMapOperation, and ProcessorRetryConfig folds onto DistributedMapProcessor. - DistributedMapError becomes DistributedMapException, matching the other 21 exception types. - Types owned by one aggregate are nested into it, removing 8 top level names. - Adds DISTRIBUTED_MAP to the exhaustive switch in ChildContextOperation. - Processor response mode is BATCH or ITEM_OUTCOMES, sending REPORT_BATCH_ITEM_OUTCOMES. The itemFailures and itemResults factories become itemOutcomes. - ItemHandlerResponse drops its two factories. - The S3 source carries a type with one matching nested config block. Transform is deleted. The seven public factory names are unchanged. - sourceExhausted and error replace completionReason and completionDetails. DistributedMapCompletionReason is deleted. - New DistributedMapError record with the four error members, shared by the summary and each item. It replaces DistributedMapResult.ItemError. - Destination include values come from two enums, one per outcome. - toleratedFailurePercentage is an integer.

feat(sdk): add distributed map operation

4237a80

zhongkechen reviewed Sep 10, 2026

View reviewed changes

nvasiu force-pushed the feat/distributed-map branch from 837e283 to f6b8990 Compare September 28, 2026 21:53

zhongkechen reviewed Oct 1, 2026

View reviewed changes

nvasiu force-pushed the feat/distributed-map branch 2 times, most recently from 74d250b to e299101 Compare October 7, 2026 23:15

nvasiu force-pushed the feat/distributed-map branch from e299101 to aca2289 Compare October 17, 2026 00:06

nvasiu force-pushed the feat/distributed-map branch from aca2289 to 2a7a4be Compare October 21, 2026 21:17

nvasiu force-pushed the feat/distributed-map branch from 2a7a4be to b5f1f51 Compare October 22, 2026 22:45

Conversation

nvasiu commented Aug 28, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Issue Link, if available

Description

Changes:

Tests

Demo/Screenshots

Checklist

Testing

Unit Tests

Integration Tests

Examples

Future Tasks

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

nvasiu commented Sep 25, 2026

Fixes

Deletions

Renames

Nesting, eight top-level names removed

Package and types

Tests

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

nvasiu commented Oct 7, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

nvasiu commented Oct 17, 2026

Model and API changes

Fixes

Validation

Documentation

Tests

Uh oh!

nvasiu commented Oct 21, 2026

Fixes

Tests

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

nvasiu commented Aug 28, 2026 •

edited

Loading

nvasiu commented Oct 7, 2026 •

edited

Loading