docs: add payload offloader examples and E2E coverage

.github/workflows/e2e-tests.yml

@@ -32,10 +32,44 @@ permissions:
   id-token: write # This is required for requesting the JWT
   contents: read # This is required for actions/checkout
 
+env:
+  AWS_REGION: us-west-2
+  FILESYSTEM_INFRASTRUCTURE_STACK_NAME: JavaSDKFileSystemPayloadE2EInfrastructureStack
+
 jobs:
+  filesystem-infrastructure:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v7
+      - name: Configure AWS credentials
+        uses: aws-actions/configure-aws-credentials@e6de054238d6b7531b4efff3b6587d9aade6a06c # v6.2.3
+        with:
+          role-to-assume: "${{ secrets.TEST_ROLE_ARN }}"
+          role-session-name: java-language-sdk-test-infrastructure
+          aws-region: ${{ env.AWS_REGION }}
+          allowed-account-ids: ${{ secrets.TEST_ACCOUNT_ID }}
+      - name: Test SAM template generator
+        run: python3 -m unittest test_generate_template.py
+        working-directory: ./examples
+      - name: Generate persistent filesystem payload E2E infrastructure template
+        run: |
+          python3 generate-template.py \
+            --file-system-infrastructure-only \
+            --output filesystem-infrastructure-template.yaml
+        working-directory: ./examples
+      - name: Ensure persistent filesystem payload E2E infrastructure
+        run: |
+          aws cloudformation deploy \
+            --template-file filesystem-infrastructure-template.yaml \
+            --stack-name ${{ env.FILESYSTEM_INFRASTRUCTURE_STACK_NAME }} \
+            --no-fail-on-empty-changeset \
+            --tags Purpose=JavaSDKFileSystemPayloadE2E
+        working-directory: ./examples
+
   e2e-tests:
+    needs: filesystem-infrastructure
     env:
-      AWS_REGION: us-west-2
       E2E_TEST_PARALLELISM: 4
     runs-on: ubuntu-latest
     strategy:
@@ -71,13 +105,24 @@ jobs:
       - name: Generate SAM template
         run: python3 generate-template.py
         working-directory: ./examples
+      - name: Generate filesystem payload E2E SAM template
+        run: python3 generate-template.py --file-system-only --output filesystem-template.yaml
+        working-directory: ./examples
       - name: sam build
         env:
           MAVEN_OPTS: -DskipTests=true -Dmaven.test.skip=true
         run: | # add --no-cached  if debugging sam build
           sam build --debug --parameter-overrides \
             'ParameterKey=Architecture,ParameterValue=x86_64 ParameterKey=JavaVersion,ParameterValue=java${{ matrix.java }} ParameterKey=FunctionNamePrefix,ParameterValue=Java${{ matrix.java }}- ParameterKey=RoleArn,ParameterValue=${{ secrets.TEST_LAMBDA_EXECUTION_ROLE_ARN }}'
         working-directory: ./examples
+      - name: sam build filesystem payload E2E stack
+        env:
+          MAVEN_OPTS: -DskipTests=true -Dmaven.test.skip=true
+        run: |
+          sam build --debug --template-file filesystem-template.yaml --build-dir .aws-sam-filesystem \
+            --parameter-overrides \
+            'ParameterKey=Architecture,ParameterValue=x86_64 ParameterKey=JavaVersion,ParameterValue=java${{ matrix.java }} ParameterKey=FunctionNamePrefix,ParameterValue=Java${{ matrix.java }}- ParameterKey=RoleArn,ParameterValue=${{ secrets.TEST_LAMBDA_EXECUTION_ROLE_ARN }} ParameterKey=FileSystemInfrastructureStackName,ParameterValue=${{ env.FILESYSTEM_INFRASTRUCTURE_STACK_NAME }}'
+        working-directory: ./examples
       - name: Clean up unmanaged Lambda log groups
         run: |
           # TODO: Remove this one-time migration cleanup after existing e2e stacks adopt managed log groups.
@@ -89,12 +134,20 @@ jobs:
             --resolve-image-repos --resolve-s3 --parameter-overrides \
             'ParameterKey=Architecture,ParameterValue=x86_64 ParameterKey=JavaVersion,ParameterValue=java${{ matrix.java }} ParameterKey=FunctionNamePrefix,ParameterValue=Java${{ matrix.java }}- ParameterKey=RoleArn,ParameterValue=${{ secrets.TEST_LAMBDA_EXECUTION_ROLE_ARN }}'
         working-directory: ./examples
+      - name: sam deploy filesystem payload E2E stack
+        run: |
+          sam deploy --template-file .aws-sam-filesystem/template.yaml \
+            --stack-name Java${{ matrix.java }}-JavaSDKFileSystemPayloadE2EStack \
+            --resolve-s3 --parameter-overrides \
+            'ParameterKey=Architecture,ParameterValue=x86_64 ParameterKey=JavaVersion,ParameterValue=java${{ matrix.java }} ParameterKey=FunctionNamePrefix,ParameterValue=Java${{ matrix.java }}- ParameterKey=RoleArn,ParameterValue=${{ secrets.TEST_LAMBDA_EXECUTION_ROLE_ARN }} ParameterKey=FileSystemInfrastructureStackName,ParameterValue=${{ env.FILESYSTEM_INFRASTRUCTURE_STACK_NAME }}'
+        working-directory: ./examples
       - name: Record E2E log start time
         run: echo "E2E_LOG_START_TIME_MS=$(date +%s%3N)" >> "$GITHUB_ENV"
       - name: Cloud Based Integration Tests
         run: |
           mvn clean test -B \
             -Dtest.cloud.enabled=true \
+            -Dtest.filesystem.enabled=true \
             -Dtest.aws.account='${{ secrets.TEST_ACCOUNT_ID }}' \
             -Dtest=CloudBasedIntegrationTest \
             -Dtest.function.name.prefix='Java${{ matrix.java }}-' \

.gitignore

@@ -42,6 +42,7 @@ __pycache__/
 # SAM
 .aws-sam/
 examples/template.yaml
+examples/filesystem-template.yaml
 samconfig.toml
 samconfig.toml.bak
 

README.md

@@ -23,6 +23,7 @@ Build resilient, long-running AWS Lambda functions that automatically checkpoint
 - **Replay Safety** – Functions deterministically resume from checkpoints after interruptions
 - **Type Safety** – Full generic type support for step results
 - **Data-Driven Concurrency** – Apply a function across a collection with `map()`, with per-item error isolation and configurable completion criteria
+- **Payload Offloading** – Keep large serialized payloads in durable external storage while checkpoints retain compact references
 
 ## How It Works
 
@@ -50,6 +51,9 @@ Your durable function extends `DurableHandler<I, O>` and implements `handleReque
 </dependency>
 ```
 
+Filesystem payload offloading is included in the core SDK artifact under
+`software.amazon.lambda.durable.offload.filesystem`.
+
 ### Your First Durable Function
 
 ```java
@@ -111,6 +115,7 @@ See [Deploy Lambda durable functions with Infrastructure as Code](https://docs.a
 **Advanced Topics**
 
 - [<u>Configuration</u>](docs/advanced/configuration.md) - Customize SDK behaviour
+- [<u>Payload Offloading</u>](docs/advanced/configuration.md#payload-offloading) - Store serialized payloads outside checkpoints
 - [<u>Error Handling</u>](docs/advanced/error-handling.md) - SDK exceptions for handling failures
 - [<u>Logging</u>](docs/advanced/logging.md) - How to use DurableLogger
 - [<u>Migrating from 1.x to 2.x</u>](docs/migration-1.x-to-2.x.md) - Upgrade guide for breaking changes since `v1.2.1`

docs/advanced/configuration.md

@@ -17,7 +17,9 @@ public class OrderProcessor extends DurableHandler<Order, OrderResult> {
         return DurableConfig.builder()
             .withLambdaClientBuilder(lambdaClientBuilder)
             .withSerDes(new MyCustomSerDes())           // Custom serialization
+            .withPayloadOffloader(myPayloadOffloader)   // Optional external payload storage
             .withExecutorService(Executors.newFixedThreadPool(10))  // Custom thread pool
+            .withPayloadOffloadExecutorService(payloadIoExecutor)    // Blocking payload I/O
             .withLoggerConfig(LoggerConfig.withReplayLogging())     // Enable replay logs
             .build();
     }
@@ -33,13 +35,103 @@ public class OrderProcessor extends DurableHandler<Order, OrderResult> {
 |-----------------------------|-----------------------------------------|-------------------------------|
 | `withLambdaClientBuilder()` | Custom AWS Lambda client                | Auto-configured Lambda client |
 | `withSerDes()`              | Serializer for step results             | Jackson with default settings |
+| `withPayloadOffloader()`    | External storage for serialized user payloads | Disabled |
 | `withExecutorService()`     | Thread pool for user-defined operations | Cached daemon thread pool     |
+| `withPayloadOffloadExecutorService()` | Thread pool for blocking payload storage I/O | Inline on the calling thread |
 | `withLoggerConfig()`        | Logger behavior configuration           | Suppress logs during replay   |
 | `withPollingStrategy()`     | Backend polling strategy                | Exponential backoff: 1s base, 2x rate, FULL jitter, 10s max |
 | `withCheckpointDelay()`     | How often the SDK checkpoints updates   | `Duration.ofSeconds(0)` (as soon as possible) |
 
 The `withExecutorService()` option configures the thread pool used for running user-defined operations. Internal SDK coordination (checkpoint batching, polling) runs on an SDK-managed thread pool.
 
+### Payload offloading
+
+`SerDes` remains responsible for converting objects to serialized text. A `PayloadOffloader` runs after serialization
+and decides whether that text remains inline or is stored externally. On replay, the SDK resolves the stored reference
+before passing the serialized text back to `SerDes`.
+
+Filesystem payload offloading is included in the core SDK artifact.
+
+Configure a durable shared mount:
+
+```java
+import java.nio.file.Path;
+import software.amazon.lambda.durable.offload.filesystem.FileSystemPathEncoding;
+import software.amazon.lambda.durable.offload.filesystem.FileSystemPayloadOffloader;
+import software.amazon.lambda.durable.offload.filesystem.PayloadOffloadMode;
+import software.amazon.lambda.durable.offload.filesystem.PreviewConfig;
+import software.amazon.lambda.durable.offload.filesystem.PreviewField;
+import software.amazon.lambda.durable.offload.filesystem.PreviewMode;
+
+var offloader = FileSystemPayloadOffloader.builder(Path.of("/mnt/efs"))
+    .storageMode(PayloadOffloadMode.OVERFLOW)
+    .pathEncoding(FileSystemPathEncoding.HASH)
+    .previewConfig(PreviewConfig.builder(PreviewMode.EXCLUDE_ALL)
+        .include(PreviewField.anywhere("id"))
+        .mask(PreviewField.anywhere("email"))
+        .build())
+    .build();
+
+return DurableConfig.builder()
+    .withSerDes(new JacksonSerDes())
+    .withPayloadOffloader(offloader)
+    .build();
+```
+
+`ALWAYS` writes every serialized payload to an immutable file. `OVERFLOW` keeps payloads inline until they approach the
+configured checkpoint-envelope limit. Every envelope records producer ownership and a SHA-256 digest; loads validate
+the owner, path, filename, and content. The configured base path and all ancestors must already exist; payload files are
+direct children of that directory. `URI` includes a bounded readable entity prefix plus an owner digest, while `HASH`
+uses only the fixed-length SHA-256 owner digest. The filesystem provider must support `SecureDirectoryStream`.
+
+The global offloader applies to root output, checkpointed step/invoke/child/map/parallel results,
+wait-for-condition state, and serialized exception data. Direct Lambda input and externally submitted callback results
+remain ordinary SerDes data. Chained invoke request payloads also remain normal Lambda JSON by default so standard
+Lambda targets do not need SDK envelope or shared-storage support. Compatible durable callers and targets can
+explicitly opt in to offloaded invoke requests. Operation configuration can override the offloader:
+
+```java
+var stepConfig = StepConfig.builder()
+    .payloadOffloader(otherOffloader)
+    .build();
+
+var inlineStepConfig = StepConfig.builder()
+    .payloadOffloader(PayloadOffloader.disabled())
+    .build();
+```
+
+The same `payloadOffloader(...)` option is available on `InvokeConfig`, `RunInChildContextConfig`, `MapConfig`,
+`ParallelConfig`, `ParallelBranchConfig`, and `WaitForConditionConfig`.
+
+Transient storage failures can be retried explicitly:
+
+```java
+var retryingOffloader = new RetryPayloadOffloader(
+    offloader,
+    RetryStrategies.fixedDelay(3, Duration.ofSeconds(1)));
+```
+
+Payload I/O runs inline by default. Configure `withPayloadOffloadExecutorService(...)` when blocking I/O should use a
+dedicated pool; it must not be the user-operation executor. Calls made from SDK-managed user-operation threads execute
+inline even when this executor is configured, preventing deadlock when separate executor wrappers share one bounded
+backing pool.
+
+The SDK uses a versioned checkpoint envelope and continues to read payloads written by older SDK versions as raw
+serialized text. Within one Lambda invocation, resolved storage data and deserialized objects use bounded weak caches
+and concurrent identical loads share one in-flight operation. Garbage collection or eviction can cause a later reload.
+
+> **Do not use Lambda `/tmp` for durable payloads.** It is local to one execution environment and might not exist on
+> replay. Use a shared durable filesystem such as EFS. S3 Files can have delayed synchronization and recent writes can
+> be lost if the runtime crashes before the mount flushes; use it only when that durability tradeoff is acceptable.
+
+The SDK does not delete offloaded files. Configure storage lifecycle and retention separately, and keep the mounted
+path accessible to every function environment that may replay or consume the payload. Treat each stored file reference
+as a capability: restrict access to the shared base path and protect checkpoint/history data containing references with
+the same controls as the payload itself.
+
+The versioned envelope, ownership, digest, filesystem, and chained-invoke contracts are defined in
+[Payload offloader wire formats](../wire-formats/payload-offloader.md).
+
 ### Dynamic plugin loading
 
 Dynamic plugin loading is an opt-in alternative to registering plugins in application code. Put provider JARs on the application class path, then set `DURABLE_EXECUTION_PLUGINS` to an ordered, comma-separated list of provider names: