feat(sdk): add distributed map operation

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,212 @@ 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.
+     *
+     * @param maxConcurrency the maximum number of items processed concurrently. Required, with no service default.
+     */
+    default <I> DistributedMapSummary distributedMap(
+            String name, DistributedMapSource<I> 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.
+     *
+     * @param maxConcurrency the maximum number of items processed concurrently. Required, with no service default.
+     */
+    default <I> DistributedMapSummary distributedMap(
+            String name,
+            DistributedMapSource<I> 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.
+     *
+     * @param maxConcurrency the maximum number of items processed concurrently. Required, with no service default.
+     */
+    default <I, O> DistributedMapResult<O> distributedMap(
+            String name,
+            DistributedMapSource<I> source,
+            DistributedMapProcessor processor,
+            int maxConcurrency,
+            TypeToken<O> 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.
+     *
+     * @param maxConcurrency the maximum number of items processed concurrently. Required, with no service default.
+     */
+    default <I, O> DistributedMapResult<O> distributedMap(
+            String name,
+            DistributedMapSource<I> source,
+            DistributedMapProcessor processor,
+            int maxConcurrency,
+            Class<O> 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.
+     *
+     * @param maxConcurrency the maximum number of items processed concurrently. Required, with no service default.
+     */
+    default <I, O> DistributedMapResult<O> distributedMap(
+            String name,
+            DistributedMapSource<I> source,
+            DistributedMapProcessor processor,
+            int maxConcurrency,
+            Class<O> 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.
+     *
+     * @param maxConcurrency the maximum number of items processed concurrently. Required, with no service default.
+     */
+    default <I, O> DistributedMapResult<O> distributedMap(
+            String name,
+            DistributedMapSource<I> source,
+            DistributedMapProcessor processor,
+            int maxConcurrency,
+            TypeToken<O> resultType) {
+        return distributedMapAsync(
+                        name,
+                        source,
+                        processor,
+                        maxConcurrency,
+                        resultType,
+                        DistributedMapConfig.builder().build())
+                .get();
+    }
+
+    /**
+     * Asynchronously runs a distributed map with default config.
+     *
+     * @param maxConcurrency the maximum number of items processed concurrently. Required, with no service default.
+     */
+    default <I> DurableFuture<DistributedMapSummary> distributedMapAsync(
+            String name, DistributedMapSource<I> source, DistributedMapProcessor processor, int maxConcurrency) {
+        return distributedMapAsync(
+                name,
+                source,
+                processor,
+                maxConcurrency,
+                DistributedMapConfig.builder().build());
+    }
+
+    /**
+     * Asynchronously runs a distributed map.
+     *
+     * @param maxConcurrency the maximum number of items processed concurrently. Required, with no service default.
+     */
+    <I> DurableFuture<DistributedMapSummary> distributedMapAsync(
+            String name,
+            DistributedMapSource<I> source,
+            DistributedMapProcessor processor,
+            int maxConcurrency,
+            DistributedMapConfig config);
+
+    /**
+     * Asynchronously runs a distributed map that collects per-item results with a TypeToken.
+     *
+     * @param maxConcurrency the maximum number of items processed concurrently. Required, with no service default.
+     */
+    <I, O> DurableFuture<DistributedMapResult<O>> distributedMapAsync(
+            String name,
+            DistributedMapSource<I> source,
+            DistributedMapProcessor processor,
+            int maxConcurrency,
+            TypeToken<O> resultType,
+            DistributedMapConfig config);
+
+    /**
+     * Asynchronously runs a distributed map that collects per-item results with a result Class.
+     *
+     * @param maxConcurrency the maximum number of items processed concurrently. Required, with no service default.
+     */
+    default <I, O> DurableFuture<DistributedMapResult<O>> distributedMapAsync(
+            String name,
+            DistributedMapSource<I> source,
+            DistributedMapProcessor processor,
+            int maxConcurrency,
+            Class<O> 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.
+     *
+     * @param maxConcurrency the maximum number of items processed concurrently. Required, with no service default.
+     */
+    default <I, O> DurableFuture<DistributedMapResult<O>> distributedMapAsync(
+            String name,
+            DistributedMapSource<I> source,
+            DistributedMapProcessor processor,
+            int maxConcurrency,
+            Class<O> 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.
+     *
+     * @param maxConcurrency the maximum number of items processed concurrently. Required, with no service default.
+     */
+    default <I, O> DurableFuture<DistributedMapResult<O>> distributedMapAsync(
+            String name,
+            DistributedMapSource<I> source,
+            DistributedMapProcessor processor,
+            int maxConcurrency,
+            TypeToken<O> 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.
      *

sdk/src/main/java/software/amazon/lambda/durable/config/DistributedMapConfig.java

@@ -0,0 +1,138 @@
+// 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 DistributedMapDestination destination;
+    private final CompletionConfig 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 DistributedMapDestination destination() {
+        return destination;
+    }
+
+    public CompletionConfig 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 DistributedMapDestination destination;
+        private CompletionConfig completionConfig;
+        private Duration timeout;
+        private SerDes resultSerDes;
+
+        private Builder() {}
+
+        public Builder destination(DistributedMapDestination destination) {
+            this.destination = destination;
+            return this;
+        }
+
+        public Builder completionConfig(CompletionConfig 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);
+        }
+    }
+
+    /** Failure-tolerance configuration for a distributed map run. */
+    public record CompletionConfig(
+            Integer toleratedFailureCount, Integer toleratedFailurePercentage, Integer minimumSampleSize) {
+
+        public CompletionConfig {
+            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");
+            }
+            // Omitting the config means zero tolerated failures, so an empty one is rejected rather than omitted.
+            if (toleratedFailureCount == null && toleratedFailurePercentage == null) {
+                throw new IllegalArgumentException(
+                        "a completion config must set toleratedFailureCount or 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);
+            }
+        }
+
+        /** Stops the run after more than this many items permanently fail. */
+        public static CompletionConfig failureCount(int count) {
+            return new CompletionConfig(count, null, null);
+        }
+
+        /** Abort once the failure rate exceeds this percentage (0 to 100). */
+        public static CompletionConfig failurePercentage(int percentage) {
+            return new CompletionConfig(null, percentage, null);
+        }
+
+        /** Abort once the failure rate exceeds this percentage (0 to 100), gated by a minimum sample size. */
+        public static CompletionConfig failurePercentage(int percentage, int minimumSampleSize) {
+            return new CompletionConfig(null, percentage, minimumSampleSize);
+        }
+    }
+}