aws · zhongkechen · Sep 2, 2026
diff --git a/sdk/src/main/java/software/amazon/lambda/durable/DurableConfig.java b/sdk/src/main/java/software/amazon/lambda/durable/DurableConfig.java
@@ -23,6 +23,7 @@
 import software.amazon.lambda.durable.client.DurableExecutionClient;
 import software.amazon.lambda.durable.client.LambdaDurableFunctionsClient;
 import software.amazon.lambda.durable.logging.LoggerConfig;
+import software.amazon.lambda.durable.offload.PayloadOffloader;
 import software.amazon.lambda.durable.plugin.DurableExecutionPlugin;
 import software.amazon.lambda.durable.plugin.PluginRunner;
 import software.amazon.lambda.durable.retry.PollingStrategies;
@@ -94,26 +95,32 @@ public final class DurableConfig {
 
     private final DurableExecutionClient durableExecutionClient;
     private final SerDes serDes;
+    private final PayloadOffloader payloadOffloader;
     private final ExecutorService executorService;
+    private final ExecutorService payloadOffloadExecutorService;
     private final LoggerConfig loggerConfig;
     private final PollingStrategy pollingStrategy;
     private final Duration checkpointDelay;
     private final boolean deserializeAfterSerialization;
     private final boolean checkpointEmptyMap;
+    private final boolean payloadOffloaderForChainedInvokePayloads;
     private final PluginRunner pluginRunner;
 
     private DurableConfig(Builder builder) {
         var plugins = DynamicPluginLoader.loadConfiguredPlugins(builder.plugins);
         this.durableExecutionClient = Objects.requireNonNullElseGet(
                 builder.durableExecutionClient, DurableConfig::createDefaultDurableExecutionClient);
         this.serDes = Objects.requireNonNullElseGet(builder.serDes, JacksonSerDes::new);
+        this.payloadOffloader = builder.payloadOffloader;
         this.executorService =
                 Objects.requireNonNullElseGet(builder.executorService, DurableConfig::createDefaultExecutor);
+        this.payloadOffloadExecutorService = builder.payloadOffloadExecutorService;
         this.loggerConfig = Objects.requireNonNullElseGet(builder.loggerConfig, LoggerConfig::defaults);
         this.pollingStrategy = Objects.requireNonNullElse(builder.pollingStrategy, PollingStrategies.Presets.DEFAULT);
         this.checkpointDelay = Objects.requireNonNullElseGet(builder.checkpointDelay, () -> Duration.ofSeconds(0));
         this.deserializeAfterSerialization = builder.deserializeAfterSerialization;
         this.checkpointEmptyMap = builder.checkpointEmptyMap;
+        this.payloadOffloaderForChainedInvokePayloads = builder.payloadOffloaderForChainedInvokePayloads;
         this.pluginRunner = plugins.isEmpty() ? PluginRunner.noOp() : new PluginRunner(plugins);
 
         validateConfiguration();
@@ -155,6 +162,11 @@ public SerDes getSerDes() {
         return serDes;
     }
 
+    /** Gets the globally configured payload offloader, or null when payload offloading is disabled. */
+    public PayloadOffloader getPayloadOffloader() {
+        return payloadOffloader;
+    }
+
     /**
      * Gets the configured ExecutorService.
      *
@@ -164,6 +176,11 @@ public ExecutorService getExecutorService() {
         return executorService;
     }
 
+    /** Gets the executor used for blocking payload offload and load operations, or null to execute inline. */
+    public ExecutorService getPayloadOffloadExecutorService() {
+        return payloadOffloadExecutorService;
+    }
+
     /**
      * Gets the configured LoggerConfig.
      *
@@ -214,6 +231,11 @@ public boolean shouldCheckpointEmptyMap() {
         return checkpointEmptyMap;
     }
 
+    /** Returns whether framed chained-invoke inputs may use the configured payload offloader. */
+    public boolean shouldUsePayloadOffloaderForChainedInvokePayloads() {
+        return payloadOffloaderForChainedInvokePayloads;
+    }
+
     /**
      * Gets the plugin runner that dispatches lifecycle events to registered plugins.
      *
@@ -235,6 +257,10 @@ public void validateConfiguration() {
         if (getExecutorService() == null) {
             throw new IllegalStateException("ExecutorService configuration failed");
         }
+        if (getPayloadOffloadExecutorService() != null && getPayloadOffloadExecutorService() == getExecutorService()) {
+            throw new IllegalStateException(
+                    "Payload offload ExecutorService must be different from the user operation ExecutorService");
+        }
     }
 
     /**
@@ -315,12 +341,15 @@ private static ExecutorService createDefaultExecutor() {
     public static final class Builder {
         private DurableExecutionClient durableExecutionClient;
         private SerDes serDes;
+        private PayloadOffloader payloadOffloader;
         private ExecutorService executorService;
+        private ExecutorService payloadOffloadExecutorService;
         private LoggerConfig loggerConfig;
         private PollingStrategy pollingStrategy;
         private Duration checkpointDelay;
         private boolean deserializeAfterSerialization = true;
         private boolean checkpointEmptyMap = false;
+        private boolean payloadOffloaderForChainedInvokePayloads;
         private List<DurableExecutionPlugin> plugins = new ArrayList<>();
 
         public Builder() {}
@@ -381,6 +410,29 @@ public Builder withSerDes(SerDes serDes) {
             return this;
         }
 
+        /**
+         * Sets the global payload offloader applied after SerDes processing.
+         *
+         * @param payloadOffloader payload offloader
+         * @return this builder
+         */
+        public Builder withPayloadOffloader(PayloadOffloader payloadOffloader) {
+            this.payloadOffloader = Objects.requireNonNull(payloadOffloader, "PayloadOffloader cannot be null");
+            return this;
+        }
+
+        /**
+         * Controls whether SDK-framed chained-invoke inputs and outputs may use this handler's configured payload
+         * offloader.
+         *
+         * <p>This is disabled by default. Enable it only for compatible durable callers that also opt in through
+         * {@link software.amazon.lambda.durable.config.InvokeConfig.Builder#usePayloadOffloaderForPayload(boolean)}.
+         */
+        public Builder withPayloadOffloaderForChainedInvokePayloads(boolean enabled) {
+            payloadOffloaderForChainedInvokePayloads = enabled;
+            return this;
+        }
+
         /**
          * Sets a custom ExecutorService for running user-defined operations. If not set, a default cached thread pool
          * will be created.
@@ -396,6 +448,22 @@ public Builder withExecutorService(ExecutorService executorService) {
             return this;
         }
 
+        /**
+         * Sets the executor used for blocking payload storage operations. If not set, payload offloader calls execute
+         * inline on the calling thread.
+         *
+         * <p>This executor must be different from the user operation executor to prevent synchronous dispatch from
+         * deadlocking a saturated operation pool.
+         *
+         * @param executorService payload offload executor
+         * @return this builder
+         */
+        public Builder withPayloadOffloadExecutorService(ExecutorService executorService) {
+            this.payloadOffloadExecutorService =
+                    Objects.requireNonNull(executorService, "Payload offload ExecutorService cannot be null");
+            return this;
+        }
+
         /**
          * Sets a custom LoggerConfig. If not set, defaults to suppressing replay logs.
          *

diff --git a/sdk/src/main/java/software/amazon/lambda/durable/config/InvokeConfig.java b/sdk/src/main/java/software/amazon/lambda/durable/config/InvokeConfig.java
@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 package software.amazon.lambda.durable.config;
 
+import software.amazon.lambda.durable.offload.PayloadOffloader;
 import software.amazon.lambda.durable.serde.SerDes;
 
 /**
@@ -12,12 +13,16 @@
 public class InvokeConfig {
     private final SerDes payloadSerDes;
     private final SerDes resultSerDes;
+    private final PayloadOffloader payloadOffloader;
     private final String tenantId;
+    private final boolean usePayloadOffloaderForPayload;
 
     public InvokeConfig(Builder builder) {
         this.payloadSerDes = builder.payloadSerDes;
         this.resultSerDes = builder.resultSerDes;
+        this.payloadOffloader = builder.payloadOffloader;
         this.tenantId = builder.tenantId;
+        this.usePayloadOffloaderForPayload = builder.usePayloadOffloaderForPayload;
     }
 
     public SerDes payloadSerDes() {
@@ -28,28 +33,43 @@ public SerDes serDes() {
         return this.resultSerDes;
     }
 
+    /** Returns the offloader used for the invoke result, or null to inherit the global offloader. */
+    public PayloadOffloader payloadOffloader() {
+        return payloadOffloader;
+    }
+
     public String tenantId() {
         return tenantId;
     }
 
+    /** Returns whether this invoke should use the framed durable-target payload protocol. */
+    public boolean usePayloadOffloaderForPayload() {
+        return usePayloadOffloaderForPayload;
+    }
+
     public static Builder builder() {
-        return new Builder(null, null, null);
+        return new Builder(null, null, null, false);
     }
 
     public Builder toBuilder() {
-        return new Builder(payloadSerDes, resultSerDes, tenantId);
+        return new Builder(payloadSerDes, resultSerDes, tenantId, usePayloadOffloaderForPayload)
+                .payloadOffloader(payloadOffloader);
     }
 
     /** Builder for creating InvokeConfig instances. */
     public static class Builder {
         private SerDes payloadSerDes;
         private SerDes resultSerDes;
+        private PayloadOffloader payloadOffloader;
         private String tenantId;
+        private boolean usePayloadOffloaderForPayload;
 
-        private Builder(SerDes payloadSerDes, SerDes resultSerDes, String tenantId) {
+        private Builder(
+                SerDes payloadSerDes, SerDes resultSerDes, String tenantId, boolean usePayloadOffloaderForPayload) {
             this.payloadSerDes = payloadSerDes;
             this.resultSerDes = resultSerDes;
             this.tenantId = tenantId;
+            this.usePayloadOffloaderForPayload = usePayloadOffloaderForPayload;
         }
 
         /**
@@ -81,6 +101,19 @@ public Builder payloadSerDes(SerDes payloadSerDes) {
             return this;
         }
 
+        /**
+         * Selects whether a compatible durable target should use the framed payload protocol for this request and its
+         * result or error.
+         *
+         * <p>This enables payload offloading for the request and lets the caller distinguish SDK-owned result/error
+         * envelopes from ordinary Lambda data. It is disabled by default so standard Lambda functions and older SDK
+         * versions continue to exchange ordinary serialized values unchanged.
+         */
+        public Builder usePayloadOffloaderForPayload(boolean enabled) {
+            usePayloadOffloaderForPayload = enabled;
+            return this;
+        }
+
         /**
          * Sets a custom serializer for the invoke result.
          *
@@ -96,6 +129,15 @@ public Builder serDes(SerDes resultSerDes) {
             return this;
         }
 
+        /**
+         * Sets the offloader for the invoke result and for the request when
+         * {@link #usePayloadOffloaderForPayload(boolean)} is enabled.
+         */
+        public Builder payloadOffloader(PayloadOffloader payloadOffloader) {
+            this.payloadOffloader = payloadOffloader;
+            return this;
+        }
+
         /**
          * Builds the InvokeConfig instance.
          *

diff --git a/sdk/src/main/java/software/amazon/lambda/durable/config/MapConfig.java b/sdk/src/main/java/software/amazon/lambda/durable/config/MapConfig.java
@@ -4,6 +4,7 @@
 
 import java.util.Objects;
 import java.util.function.BiFunction;
+import software.amazon.lambda.durable.offload.PayloadOffloader;
 import software.amazon.lambda.durable.serde.SerDes;
 
 /**
@@ -15,6 +16,7 @@ public class MapConfig {
     private final Integer maxConcurrency;
     private final CompletionConfig completionConfig;
     private final SerDes serDes;
+    private final PayloadOffloader payloadOffloader;
     private final NestingType nestingType;
     private final BiFunction<Object, Integer, String> itemNamer;
 
@@ -23,6 +25,7 @@ private MapConfig(Builder builder) {
         this.completionConfig = Objects.requireNonNullElse(builder.completionConfig, CompletionConfig.allCompleted());
         this.nestingType = Objects.requireNonNullElse(builder.nestingType, NestingType.NESTED);
         this.serDes = builder.serDes;
+        this.payloadOffloader = builder.payloadOffloader;
         this.itemNamer = builder.itemNamer;
         if (itemNamer != null && nestingType == NestingType.FLAT) {
             throw new IllegalArgumentException("itemNamer is not supported with FLAT map nesting");
@@ -44,6 +47,11 @@ public SerDes serDes() {
         return serDes;
     }
 
+    /** @return the map and iteration result offloader, or null to inherit the global offloader */
+    public PayloadOffloader payloadOffloader() {
+        return payloadOffloader;
+    }
+
     /** @return nesting type, defaults to {@link NestingType#NESTED} */
     public NestingType nestingType() {
         return nestingType;
@@ -70,6 +78,7 @@ public Builder toBuilder() {
                 .maxConcurrency(maxConcurrency)
                 .completionConfig(completionConfig)
                 .serDes(serDes)
+                .payloadOffloader(payloadOffloader)
                 .nestingType(nestingType)
                 .itemNamer(itemNamer);
     }
@@ -80,6 +89,7 @@ public static class Builder {
         private Integer maxConcurrency;
         private CompletionConfig completionConfig;
         private SerDes serDes;
+        private PayloadOffloader payloadOffloader;
         private BiFunction<Object, Integer, String> itemNamer;
 
         private Builder() {}
@@ -114,6 +124,12 @@ public Builder serDes(SerDes serDes) {
             return this;
         }
 
+        /** Sets the payload offloader for map iteration and aggregate results. */
+        public Builder payloadOffloader(PayloadOffloader payloadOffloader) {
+            this.payloadOffloader = payloadOffloader;
+            return this;
+        }
+
         /**
          * Sets the nesting type for the map operation.
          *

diff --git a/sdk/src/main/java/software/amazon/lambda/durable/config/ParallelBranchConfig.java b/sdk/src/main/java/software/amazon/lambda/durable/config/ParallelBranchConfig.java
@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 package software.amazon.lambda.durable.config;
 
+import software.amazon.lambda.durable.offload.PayloadOffloader;
 import software.amazon.lambda.durable.serde.SerDes;
 
 /**
@@ -11,18 +12,25 @@
  */
 public class ParallelBranchConfig {
     private final SerDes serDes;
+    private final PayloadOffloader payloadOffloader;
 
     private ParallelBranchConfig(Builder builder) {
         this.serDes = builder.serDes;
+        this.payloadOffloader = builder.payloadOffloader;
     }
 
     /** Returns the custom serializer for this step, or null if not specified (uses default SerDes). */
     public SerDes serDes() {
         return serDes;
     }
 
+    /** Returns the branch result offloader, or null to inherit the global offloader. */
+    public PayloadOffloader payloadOffloader() {
+        return payloadOffloader;
+    }
+
     public Builder toBuilder() {
-        return new Builder(serDes);
+        return new Builder(serDes).payloadOffloader(payloadOffloader);
     }
 
     /**
@@ -37,6 +45,7 @@ public static Builder builder() {
     /** Builder for creating StepConfig instances. */
     public static class Builder {
         private SerDes serDes;
+        private PayloadOffloader payloadOffloader;
 
         public Builder(SerDes serDes) {
             this.serDes = serDes;
@@ -57,6 +66,12 @@ public Builder serDes(SerDes serDes) {
             return this;
         }
 
+        /** Sets the payload offloader for the parallel branch result. */
+        public Builder payloadOffloader(PayloadOffloader payloadOffloader) {
+            this.payloadOffloader = payloadOffloader;
+            return this;
+        }
+
         /**
          * Builds the ParallelBranchConfig instance.
          *