Merge branch 'main' into phipag/error-handling

phipag · phipag · commit f802fa4ca134 · 2026-01-27T12:42:08.000+01:00
diff --git a/.github/workflows/sync-package.yml b/.github/workflows/sync-package.yml
@@ -0,0 +1,76 @@
+name: Sync package
+
+env:
+  AWS_REGION: "us-west-2"
+
+# permission can be added at job level or workflow level
+permissions:
+  id-token: write # This is required for requesting the JWT
+  contents: read # This is required for actions/checkout
+
+on:
+  workflow_call:
+    inputs:
+      module_name:
+        required: true
+        type: string
+      package_name:
+        required: true
+        type: string
+    secrets:
+      S3_BUCKET_NAME:
+        required: true
+      SYNC_LAMBDA_ARN:
+        required: true
+      GITFARM_LAN_SDK_REPO:
+        required: true
+      GITFARM_LAN_SDK_BRANCH:
+        required: true
+      ACTIONS_SYNC_ROLE_NAME:
+        required: true
+jobs:
+  upload-to-S3-and-sync-to-Gitfarm:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Git clone the repository
+        uses: actions/checkout@v5
+      - name: Set up JDK 17
+        uses: actions/setup-java@v4
+        with:
+          java-version: '17'
+          distribution: 'temurin'
+          cache: maven
+      - name: Get version
+        id: get_version
+        run: |
+          VERSION=$( mvn -f ${{ inputs.module_name }}/pom.xml help:evaluate -Dexpression=project.version -q -DforceStdout )
+          echo "version=$VERSION" >> "$GITHUB_OUTPUT"
+      - name: Show extracted Maven project version
+        run: echo ${{ steps.get_version.outputs.version }}
+      - name: Build with Maven
+        run: mvn -B package --file pom.xml
+      - name: configure aws credentials
+        uses: aws-actions/configure-aws-credentials@e3dd6a429d7300a6a4c196c26e071d42e0343502
+        with:
+          role-to-assume: "${{ secrets.ACTIONS_SYNC_ROLE_NAME }}"
+          role-session-name: samplerolesession
+          aws-region: ${{ env.AWS_REGION }}
+      # Upload a file to AWS s3
+      - name: Copy tgz build file to s3
+        run: |
+          aws s3 cp ./${{ inputs.module_name }}/target/${{ inputs.package_name }}-${{ steps.get_version.outputs.version }}.jar \
+            s3://${{ secrets.S3_BUCKET_NAME }}/${{ inputs.package_name }}.jar
+      - name: commit to Gitfarm
+        run: |
+          aws lambda invoke \
+            --function-name ${{ secrets.SYNC_LAMBDA_ARN }} \
+            --payload '{"gitFarmRepo":"${{ secrets.GITFARM_LAN_SDK_REPO }}","gitFarmBranch":"${{ secrets.GITFARM_LAN_SDK_BRANCH }}","gitFarmFilepath":"${{ secrets.GITFARM_LAN_SDK_REPO }}-1.0.jar","s3Bucket":"${{ secrets.S3_BUCKET_NAME }}","s3FilePath":"${{ inputs.package_name }}.jar", "gitHubRepo": "aws-durable-execution-sdk-java", "gitHubCommit":"${{ github.sha }}"}' \
+            --cli-binary-format raw-in-base64-out \
+            output.txt
+      - name: Check for specific text in a file
+        id: check_text
+        run: |
+          if grep -q "Error" output.txt; then
+            cat output.txt
+            exit 1
+          fi
diff --git a/.github/workflows/sync-sdk.yml b/.github/workflows/sync-sdk.yml
@@ -0,0 +1,27 @@
+name: Sync SDK
+on:
+  push:
+    branches: ["main"]
+    paths:
+      - "sdk/**"
+
+env:
+  AWS_REGION: "us-west-2"
+
+# permission can be added at job level or workflow level
+permissions:
+  id-token: write # This is required for requesting the JWT
+  contents: read # This is required for actions/checkout
+
+jobs:
+  sync-aws-durable-execution-sdk-java:
+    uses: ./.github/workflows/sync-package.yml
+    with:
+      module_name: sdk
+      package_name: aws-durable-execution-sdk-java
+    secrets:
+      S3_BUCKET_NAME: ${{ secrets.S3_BUCKET_NAME }}
+      SYNC_LAMBDA_ARN: ${{ secrets.SYNC_LAMBDA_ARN }}
+      GITFARM_LAN_SDK_REPO: ${{ secrets.GITFARM_LAN_SDK_REPO }}
+      GITFARM_LAN_SDK_BRANCH: ${{ secrets.GITFARM_LAN_SDK_BRANCH }}
+      ACTIONS_SYNC_ROLE_NAME: ${{ secrets.ACTIONS_SYNC_ROLE_NAME }}
diff --git a/.github/workflows/sync-testing-sdk.yml b/.github/workflows/sync-testing-sdk.yml
@@ -0,0 +1,28 @@
+name: Sync Testing SDK
+
+on:
+  push:
+    branches: ["main"]
+    paths:
+      - "sdk-testing/**"
+
+env:
+  AWS_REGION: "us-west-2"
+
+# permission can be added at job level or workflow level
+permissions:
+  id-token: write # This is required for requesting the JWT
+  contents: read # This is required for actions/checkout
+
+jobs:
+  sync-aws-durable-execution-sdk-java-testing:
+    uses: ./.github/workflows/sync-package.yml
+    with:
+      module_name: sdk-testing
+      package_name: aws-durable-execution-sdk-java-testing
+    secrets:
+      S3_BUCKET_NAME: ${{ secrets.S3_BUCKET_NAME }}
+      SYNC_LAMBDA_ARN: ${{ secrets.SYNC_LAMBDA_ARN }}
+      GITFARM_LAN_SDK_REPO: ${{ secrets.GITFARM_TESTING_SDK_REPO }}
+      GITFARM_LAN_SDK_BRANCH: ${{ secrets.GITFARM_LAN_SDK_BRANCH }}
+      ACTIONS_SYNC_ROLE_NAME: ${{ secrets.ACTIONS_SYNC_ROLE_NAME }}
diff --git a/examples/src/main/java/com/amazonaws/lambda/durable/examples/CustomConfigExample.java b/examples/src/main/java/com/amazonaws/lambda/durable/examples/CustomConfigExample.java
@@ -123,15 +123,6 @@ public String serialize(Object obj) {
             }
         }
 
-        @Override
-        public <T> T deserialize(String json, Class<T> clazz) {
-            try {
-                return objectMapper.readValue(json, clazz);
-            } catch (IOException e) {
-                throw new RuntimeException("Deserialization failed", e);
-            }
-        }
-
         @Override
         public <T> T deserialize(String json, TypeToken<T> typeToken) {
             try {
diff --git a/sdk-integration-tests/src/test/java/com/amazonaws/lambda/durable/CustomConfigIntegrationTest.java b/sdk-integration-tests/src/test/java/com/amazonaws/lambda/durable/CustomConfigIntegrationTest.java
@@ -34,12 +34,6 @@ public String serialize(Object value) {
             return delegate.serialize(value);
         }
 
-        @Override
-        public <T> T deserialize(String data, Class<T> type) {
-            deserializeCount.incrementAndGet();
-            return delegate.deserialize(data, type);
-        }
-
         @Override
         public <T> T deserialize(String data, TypeToken<T> typeToken) {
             deserializeCount.incrementAndGet();
diff --git a/sdk-integration-tests/src/test/java/com/amazonaws/lambda/durable/CustomSerDesIntegrationTest.java b/sdk-integration-tests/src/test/java/com/amazonaws/lambda/durable/CustomSerDesIntegrationTest.java
@@ -26,12 +26,6 @@ public String serialize(Object value) {
             return delegate.serialize(value);
         }
 
-        @Override
-        public <T> T deserialize(String data, Class<T> type) {
-            deserializeCount.incrementAndGet();
-            return delegate.deserialize(data, type);
-        }
-
         @Override
         public <T> T deserialize(String data, TypeToken<T> typeToken) {
             deserializeCount.incrementAndGet();
diff --git a/sdk-testing/src/main/java/com/amazonaws/lambda/durable/testing/TestOperation.java b/sdk-testing/src/main/java/com/amazonaws/lambda/durable/testing/TestOperation.java
@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 package com.amazonaws.lambda.durable.testing;
 
+import com.amazonaws.lambda.durable.TypeToken;
 import com.amazonaws.lambda.durable.serde.SerDes;
 import java.util.List;
 import software.amazon.awssdk.services.lambda.model.ErrorObject;
@@ -58,7 +59,7 @@ public <T> T getStepResult(Class<T> type) {
         if (details == null || details.result() == null) {
             return null;
         }
-        return serDes.deserialize(details.result(), type);
+        return serDes.deserialize(details.result(), TypeToken.get(type));
     }
 
     public ErrorObject getError() {
diff --git a/sdk-testing/src/main/java/com/amazonaws/lambda/durable/testing/TestResult.java b/sdk-testing/src/main/java/com/amazonaws/lambda/durable/testing/TestResult.java
@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 package com.amazonaws.lambda.durable.testing;
 
+import com.amazonaws.lambda.durable.TypeToken;
 import com.amazonaws.lambda.durable.model.ExecutionStatus;
 import com.amazonaws.lambda.durable.serde.SerDes;
 import java.util.List;
@@ -49,7 +50,7 @@ public <T> T getResult(Class<T> resultType) {
         if (resultPayload == null) {
             return null;
         }
-        return serDes.deserialize(resultPayload, resultType);
+        return serDes.deserialize(resultPayload, TypeToken.get(resultType));
     }
 
     public Optional<ErrorObject> getError() {
diff --git a/sdk/src/main/java/com/amazonaws/lambda/durable/DurableContext.java b/sdk/src/main/java/com/amazonaws/lambda/durable/DurableContext.java
@@ -56,7 +56,7 @@ public class DurableContext {
     public <T> T step(String name, Class<T> resultType, Supplier<T> func) {
         return step(
                 name,
-                resultType,
+                TypeToken.get(resultType),
                 func,
                 StepConfig.builder()
                         .retryStrategy(RetryStrategies.Presets.NO_RETRY)
@@ -86,29 +86,15 @@ public <T> T step(String name, TypeToken<T> typeToken, Supplier<T> func, StepCon
     public <T> DurableFuture<T> stepAsync(String name, Class<T> resultType, Supplier<T> func) {
         return stepAsync(
                 name,
-                resultType,
+                TypeToken.get(resultType),
                 func,
                 StepConfig.builder()
                         .retryStrategy(RetryStrategies.Presets.NO_RETRY)
                         .build());
     }
 
     public <T> DurableFuture<T> stepAsync(String name, Class<T> resultType, Supplier<T> func, StepConfig config) {
-        var operationId = nextOperationId();
-
-        // Validate replay consistency
-        var existing = executionManager.getOperation(operationId);
-        if (existing != null) {
-            validateReplay(operationId, OperationType.STEP, name, existing);
-        }
-
-        // Create and start step operation
-        StepOperation<T> operation =
-                new StepOperation<>(operationId, name, func, resultType, config, executionManager, logger, serDes);
-
-        operation.execute(); // Start the step (returns immediately)
-
-        return new DurableFuture<>(operation);
+        return stepAsync(name, TypeToken.get(resultType), func, config);
     }
 
     public <T> DurableFuture<T> stepAsync(String name, TypeToken<T> typeToken, Supplier<T> func) {
diff --git a/sdk/src/main/java/com/amazonaws/lambda/durable/DurableExecutor.java b/sdk/src/main/java/com/amazonaws/lambda/durable/DurableExecutor.java
@@ -156,7 +156,7 @@ private static <I> I extractUserInput(Operation executionOp, SerDes serDes, Clas
         }
 
         var inputPayload = executionOp.executionDetails().inputPayload();
-        return serDes.deserialize(inputPayload, inputType);
+        return serDes.deserialize(inputPayload, TypeToken.get(inputType));
     }
 
     public static <I, O> RequestHandler<DurableExecutionInput, DurableExecutionOutput> wrap(
diff --git a/sdk/src/main/java/com/amazonaws/lambda/durable/TypeToken.java b/sdk/src/main/java/com/amazonaws/lambda/durable/TypeToken.java
@@ -44,6 +44,14 @@ protected TypeToken() {
         }
     }
 
+    private TypeToken(Type type) {
+        this.type = type;
+    }
+
+    public static <U> TypeToken<U> get(Class<U> clazz) {
+        return new TypeToken<>(clazz) {};
+    }
+
     /**
      * Returns the captured type.
      *
diff --git a/sdk/src/main/java/com/amazonaws/lambda/durable/operation/StepOperation.java b/sdk/src/main/java/com/amazonaws/lambda/durable/operation/StepOperation.java
@@ -35,26 +35,13 @@ public class StepOperation<T> implements DurableOperation<T> {
     private final String operationId;
     private final String name;
     private final Supplier<T> function;
-    private final Class<T> resultType;
     private final TypeToken<T> resultTypeToken;
     private final StepConfig config;
     private final ExecutionManager executionManager;
     private final DurableLogger durableLogger;
     private final SerDes serDes;
     private final Phaser phaser;
 
-    public StepOperation(
-            String operationId,
-            String name,
-            Supplier<T> function,
-            Class<T> resultType,
-            StepConfig config,
-            ExecutionManager executionManager,
-            DurableLogger durableLogger,
-            SerDes serDes) {
-        this(operationId, name, function, resultType, null, config, executionManager, durableLogger, serDes);
-    }
-
     public StepOperation(
             String operationId,
             String name,
@@ -64,30 +51,13 @@ public StepOperation(
             ExecutionManager executionManager,
             DurableLogger durableLogger,
             SerDes serDes) {
-        this(operationId, name, function, null, resultTypeToken, config, executionManager, durableLogger, serDes);
-    }
-
-    private StepOperation(
-            String operationId,
-            String name,
-            Supplier<T> function,
-            Class<T> resultType,
-            TypeToken<T> resultTypeToken,
-            StepConfig config,
-            ExecutionManager executionManager,
-            DurableLogger durableLogger,
-            SerDes serDes) {
-        if (resultType == null && resultTypeToken == null) {
+        if (resultTypeToken == null) {
             throw new IllegalArgumentException("Either resultType or resultTypeToken must be provided");
         }
-        if (resultType != null && resultTypeToken != null) {
-            throw new IllegalArgumentException("Cannot provide both resultType and resultTypeToken");
-        }
 
         this.operationId = operationId;
         this.name = name;
         this.function = function;
-        this.resultType = resultType;
         this.resultTypeToken = resultTypeToken;
         this.config = config;
         this.executionManager = executionManager;
@@ -339,12 +309,7 @@ public T get() {
             var stepDetails = op.stepDetails();
             var result = (stepDetails != null) ? stepDetails.result() : null;
 
-            // Use TypeToken if provided, otherwise use Class
-            if (resultTypeToken != null) {
-                return serDes.deserialize(result, resultTypeToken);
-            } else {
-                return serDes.deserialize(result, resultType);
-            }
+            return serDes.deserialize(result, resultTypeToken);
         } else {
             var errorType = op.stepDetails().error().errorType();
 
@@ -357,8 +322,9 @@ public T get() {
             try {
                 Class<?> exceptionClass = Class.forName(errorType);
                 if (Throwable.class.isAssignableFrom(exceptionClass)) {
-                    Throwable original = (Throwable) serDes.deserialize(
-                            op.stepDetails().error().errorData(), exceptionClass.asSubclass(Throwable.class));
+                    Throwable original = serDes.deserialize(
+                            op.stepDetails().error().errorData(),
+                            TypeToken.get(exceptionClass.asSubclass(Throwable.class)));
 
                     if (original != null) {
                         original.setStackTrace(StepFailedException.deserializeStackTrace(
diff --git a/sdk/src/main/java/com/amazonaws/lambda/durable/serde/JacksonSerDes.java b/sdk/src/main/java/com/amazonaws/lambda/durable/serde/JacksonSerDes.java
@@ -54,16 +54,6 @@ public String serialize(Object value) {
         }
     }
 
-    @Override
-    public <T> T deserialize(String data, Class<T> type) {
-        if (data == null) return null;
-        try {
-            return mapper.readValue(data, type);
-        } catch (Exception e) {
-            throw new SerDesException("Deserialization failed for type: " + type.getName(), e);
-        }
-    }
-
     @Override
     public <T> T deserialize(String data, TypeToken<T> typeToken) {
         if (data == null) return null;
diff --git a/sdk/src/main/java/com/amazonaws/lambda/durable/serde/SerDes.java b/sdk/src/main/java/com/amazonaws/lambda/durable/serde/SerDes.java
@@ -18,16 +18,6 @@ public interface SerDes {
      */
     String serialize(Object value);
 
-    /**
-     * Deserializes a JSON string to an object of the specified type.
-     *
-     * @param data the JSON string to deserialize
-     * @param type the class of the target type
-     * @param <T> the target type
-     * @return the deserialized object, or null if data is null
-     */
-    <T> T deserialize(String data, Class<T> type);
-
     /**
      * Deserializes a JSON string to an object of the specified generic type.
      *
diff --git a/sdk/src/test/java/com/amazonaws/lambda/durable/DurableExecutionWrapperTest.java b/sdk/src/test/java/com/amazonaws/lambda/durable/DurableExecutionWrapperTest.java
diff --git a/sdk/src/test/java/com/amazonaws/lambda/durable/operation/StepOperationTest.java b/sdk/src/test/java/com/amazonaws/lambda/durable/operation/StepOperationTest.java
diff --git a/sdk/src/test/java/com/amazonaws/lambda/durable/serde/JacksonSerDesTest.java b/sdk/src/test/java/com/amazonaws/lambda/durable/serde/JacksonSerDesTest.java

Original file line number	Diff line number	Diff line change
`@@ -2,6 +2,7 @@`
`2`	`2`	`// SPDX-License-Identifier: Apache-2.0`
`3`	`3`	`package com.amazonaws.lambda.durable.testing;`
`4`	`4`
	`5`	`+import com.amazonaws.lambda.durable.TypeToken;`
`5`	`6`	`import com.amazonaws.lambda.durable.serde.SerDes;`
`6`	`7`	`import java.util.List;`
`7`	`8`	`import software.amazon.awssdk.services.lambda.model.ErrorObject;`
`@@ -58,7 +59,7 @@ public <T> T getStepResult(Class<T> type) {`
`58`	`59`	`if (details == null \|\| details.result() == null) {`
`59`	`60`	`return null;`
`60`	`61`	`}`
`61`		`- return serDes.deserialize(details.result(), type);`
	`62`	`+ return serDes.deserialize(details.result(), TypeToken.get(type));`
`62`	`63`	`}`
`63`	`64`
`64`	`65`	`public ErrorObject getError() {`
Original file line number	Diff line number	Diff line change
`@@ -156,7 +156,7 @@ private static <I> I extractUserInput(Operation executionOp, SerDes serDes, Clas`
`156`	`156`	`}`
`157`	`157`
`158`	`158`	`var inputPayload = executionOp.executionDetails().inputPayload();`
`159`		`- return serDes.deserialize(inputPayload, inputType);`
	`159`	`+ return serDes.deserialize(inputPayload, TypeToken.get(inputType));`
`160`	`160`	`}`
`161`	`161`
`162`	`162`	`public static <I, O> RequestHandler<DurableExecutionInput, DurableExecutionOutput> wrap(`