Repository navigation
Conversation
|
Refactored to address @zhongkechen comments. The following changes were made: Fixes
Deletions
Renames
Nesting, eight top-level names removed
Package and types
Tests
|
837e283 to
f6b8990
Compare
| var index = i; | ||
| futures.add(pool.submit(() -> { | ||
| I item = toItem(inSerdes, records.get(index).body(), itemType); | ||
| outputs[index] = outSerdes.serialize(func.apply(item)); |
There was a problem hiding this comment.
Is the output used for failed items?
There was a problem hiding this comment.
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.
| return (event, context) -> { | ||
| S state = event.state() != null ? serdes.deserialize(event.state(), stateType) : null; | ||
|
|
||
| var page = func.apply(state); |
There was a problem hiding this comment.
how would the reader know maxItems?
There was a problem hiding this comment.
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.
| for (var record : records) { | ||
| bodies.add(record.body()); | ||
| } | ||
| MapResult<String> batch = ctx.map( |
There was a problem hiding this comment.
How many threads would be created at maximum for this map operation?
There was a problem hiding this comment.
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).
| @Override | ||
| protected void start() { | ||
| sendOperationUpdate( | ||
| OperationUpdate.builder().action(OperationAction.START).distributedMapOptions(options)); |
There was a problem hiding this comment.
Is it possible to exceed the checkpoint size limit here?
There was a problem hiding this comment.
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.
|
New revision to address recent comments:
Other changes:
|
f6b8990 to
74d250b
Compare
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. 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.
74d250b to
e299101
Compare
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/The result types a customer receives back from a map run.
DistributedMapSummary: whatctx.distributedMapreturns, describes the run's overall outcome.throwIfError()opts into raising.DistributedMapResult: returned when a result type is passed. Contains individual map run item outcomes.DistributedMapResult.ItemandDistributedMapResult.ItemError: represent a single item's result / error. Nested, since each belongs to exactly one result.DistributedMapStatusandDistributedMapCompletionReason: the run enums.config/The input types a customer constructs to describe a map run.
DistributedMapConfig: optional settings for distributed map, nestingCompletionConfigfor the item failure thresholds that mark the overall run failed.DistributedMapSource: describes where map run items come from, with factories for inline, S3 and reader sources, and nesting the CSV options.DistributedMapProcessor: describes the Lambda that processes items, how outcomes are reported back viabatch,itemFailuresanditemResults, and how failing items are retried viamaxRetryAttemptsandmaxRetryDuration.DistributedMapDestination: routes successful and failed item records to S3, nestingSuccessandFailureso a success destination cannot be used for failures.DurableContext.javaandcontext/DurableContextImpl.javaThe 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 isDistributedMapResultwhen a result type is passed andDistributedMapSummaryotherwise, withClass<O>andTypeToken<O>forms, blocking and async forms, and optional config.maxConcurrencyagainst the model range of 1 to 10000.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 variantscreateDistributedMapItemHandlerWithDurableExecutionandcreateDistributedMapBatchHandlerWithDurableExecution.concurrencyhandlers at a time, defaulting to one.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.javaThe executor that drives the operation against the durable execution runtime.
throwIfError()opts into raising.DistributedMapOptionsand parsingDistributedMapDetailsback.Items,Outputand the record body are opaque strings, matching the API model, so any serdes works.operation/ChildContextOperation.javaOperation type handling.
DISTRIBUTED_MAPin the operation sub-type switch.exception/DistributedMapException.javaThe error type a customer catches.
util/DistributedMapValidation.javaShared validation and S3 URI parsing.
execution/CheckpointManager.javaBatching 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/DistributedMapConfigTest(11),DistributedMapSourceTest(36),DistributedMapProcessorTest(17),DistributedMapDestinationTest(14).sdk/src/test/.../model/DistributedMapResultTest(11),DistributedMapSummaryTest(7),DistributedMapStatusTest(4),DistributedMapCompletionReasonTest(4).sdk/src/test/.../operation/DistributedMapOperationTest(8),DistributedMapOperationTranslationTest(47), which asserts the wire shape in both directions.sdk/src/test/.../context/DurableContextDistributedMapTest.java(9)ctx.distributedMapsurface and its validation.sdk/src/test/.../dmap/DistributedMapHandlersTest.java(24)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
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