feat(sdk): Treat KMS exceptions as non-retryable customer errors

KMS exceptions from Lambda (e.g., KMSAccessDeniedException,
KMSDisabledException) arrive as 502 errors during
CheckpointDurableExecution and GetDurableExecutionState API calls.
Unlike other 5xx errors, we do not want Durable Functions to retry
invocations for these errors, as they would never self-resolve.

Add DurableApiErrorClassifier in the exception package with a
NON_RETRYABLE_ERROR_CODES set and classifyException method. Use it
in CheckpointManager to wrap both checkpoint() and
getExecutionState() calls so that KMS errors return
UnrecoverableDurableExecutionException with status FAILED
immediately. The classifier also handles 4xx (non-429) as
non-retryable and Invalid Checkpoint Token as retryable, matching
the JS SDK's classifyCheckpointError logic.

The classifyException method could be further updated to return a
new UnrecoverableDurableInvocationException for retryable Durable
API errors (e.g., 429, 5xx), allowing the SDK to distinguish them
from user code errors and crash the invocation for retry instead
of misclassifying them as step failures.

Testing — parameterized unit tests for all four KMS exceptions
with realistic error messages, plus tests for 4xx, Invalid
Checkpoint Token, non-token InvalidParameterValueException, 429,
500, non-KMS 502, error detail preservation, and two wiring tests
verifying CheckpointManager delegates to the classifier for both
API paths.

By submitting this pull request, I confirm that you can use,
modify, copy, and redistribute this contribution, under the terms
of your choice.

Author: denzyll

sdk/src/main/java/software/amazon/lambda/durable/exception/DurableApiErrorClassifier.java

@@ -0,0 +1,111 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+package software.amazon.lambda.durable.exception;
+
+import java.util.Set;
+import software.amazon.awssdk.awscore.exception.AwsServiceException;
+import software.amazon.awssdk.services.lambda.model.ErrorObject;
+
+/**
+ * Classifies AWS service exceptions from Durable Execution API calls as execution-level (non-retryable) or
+ * invocation-level (retryable).
+ *
+ * <p>Execution-level errors throw {@link UnrecoverableDurableExecutionException} to terminate the execution
+ * immediately. These represent permanent customer-side issues (e.g., KMS key misconfiguration) that will not
+ * self-resolve on retry.
+ *
+ * <p>Invocation-level errors are allowed to propagate, crashing the current Lambda invocation so the backend can retry
+ * with a fresh invocation.
+ *
+ * <p>To add a new non-retryable error, add its error code to {@link #NON_RETRYABLE_ERROR_CODES}.
+ */
+public final class DurableApiErrorClassifier {
+
+    /**
+     * Error codes that represent non-retryable customer errors. When a Durable Execution API call fails with one of
+     * these error codes, the execution is terminated immediately.
+     *
+     * <p>These error codes are documented under the Lambda Invoke API but also apply to other Lambda APIs such as
+     * {@code CheckpointDurableExecution} and {@code GetDurableExecutionState}.
+     *
+     * @see <a href="https://docs.aws.amazon.com/lambda/latest/api/API_Invoke.html">Lambda Invoke API — Errors
+     *     (reference for KMS exception names)</a>
+     */
+    static final Set<String> NON_RETRYABLE_ERROR_CODES = Set.of(
+            "KMSAccessDeniedException", "KMSDisabledException", "KMSInvalidStateException", "KMSNotFoundException");
+
+    /** HTTP 429 (Too Many Requests) indicates throttling — a transient condition that resolves on retry. */
+    private static final int THROTTLING_STATUS_CODE = 429;
+
+    /**
+     * Error code for invalid checkpoint token errors. These occur when the SDK uses a stale checkpoint token that has
+     * been superseded by a newer invocation. Retrying with a fresh invocation resolves this.
+     *
+     * @see <a
+     *     href="https://github.com/aws/aws-durable-execution-sdk-js/blob/main/packages/aws-durable-execution-sdk-js/src/utils/checkpoint/checkpoint-manager.ts">JS
+     *     SDK classifyCheckpointError</a>
+     */
+    private static final String INVALID_CHECKPOINT_TOKEN_ERROR_CODE = "InvalidParameterValueException";
+
+    /**
+     * Message prefix that distinguishes invalid checkpoint token errors from other
+     * {@code InvalidParameterValueException} errors.
+     */
+    private static final String INVALID_CHECKPOINT_TOKEN_MESSAGE_PREFIX = "Invalid Checkpoint Token";
+
+    private DurableApiErrorClassifier() {}
+
+    /**
+     * Classifies the given exception and returns the appropriate exception to throw.
+     *
+     * <p>Returns {@link UnrecoverableDurableExecutionException} for non-retryable customer errors, or the original
+     * exception for retryable errors.
+     *
+     * <p>Classification rules:
+     *
+     * <ul>
+     *   <li>Error code in {@link #NON_RETRYABLE_ERROR_CODES} → execution error (non-retryable)
+     *   <li>4xx + "Invalid Checkpoint Token" → invocation error (retryable, stale token resolves on retry)
+     *   <li>4xx (non-429) → execution error (non-retryable customer error)
+     *   <li>429, 5xx, unknown → invocation error (retryable)
+     * </ul>
+     *
+     * @param e the AWS service exception from a Durable Execution API call
+     * @return an {@link UnrecoverableDurableExecutionException} if non-retryable, or the original exception if
+     *     retryable
+     */
+    public static RuntimeException classifyException(AwsServiceException e) {
+        var errorCode = e.awsErrorDetails().errorCode();
+
+        // Non-retryable customer errors: execution is terminally broken (e.g., KMS key misconfiguration)
+        if (NON_RETRYABLE_ERROR_CODES.contains(errorCode)) {
+            return buildUnrecoverableDurableExecutionException(e);
+        }
+
+        var statusCode = e.awsErrorDetails().sdkHttpResponse().statusCode();
+        var message = e.getMessage();
+
+        // 4xx errors (excluding throttling) are non-retryable customer errors
+        if (statusCode >= 400 && statusCode < 500 && statusCode != THROTTLING_STATUS_CODE) {
+            // Stale checkpoint token: occurs when a newer invocation has superseded this one.
+            // Retrying with a fresh invocation resolves this, so treat as retryable.
+            if (INVALID_CHECKPOINT_TOKEN_ERROR_CODE.equals(errorCode)
+                    && message != null
+                    && message.startsWith(INVALID_CHECKPOINT_TOKEN_MESSAGE_PREFIX)) {
+                return e;
+            }
+            return buildUnrecoverableDurableExecutionException(e);
+        }
+
+        // 429 (throttling), 5xx (service errors), unknown — transient, retryable
+        return e;
+    }
+
+    private static UnrecoverableDurableExecutionException buildUnrecoverableDurableExecutionException(
+            AwsServiceException e) {
+        return new UnrecoverableDurableExecutionException(ErrorObject.builder()
+                .errorType(e.awsErrorDetails().errorCode())
+                .errorMessage(e.getMessage())
+                .build());
+    }
+}

sdk/src/main/java/software/amazon/lambda/durable/execution/CheckpointManager.java

@@ -14,10 +14,12 @@
 import java.util.function.Consumer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import software.amazon.awssdk.awscore.exception.AwsServiceException;
 import software.amazon.awssdk.services.lambda.model.CheckpointUpdatedExecutionState;
 import software.amazon.awssdk.services.lambda.model.Operation;
 import software.amazon.awssdk.services.lambda.model.OperationUpdate;
 import software.amazon.lambda.durable.DurableConfig;
+import software.amazon.lambda.durable.exception.DurableApiErrorClassifier;
 import software.amazon.lambda.durable.retry.PollingStrategies;
 import software.amazon.lambda.durable.retry.PollingStrategy;
 
@@ -163,14 +165,18 @@ List<Operation> fetchAllPages(CheckpointUpdatedExecutionState checkpointUpdatedE
         var nextMarker = checkpointUpdatedExecutionState.nextMarker();
         while (nextMarker != null && !nextMarker.isEmpty()) {
             var startTime = System.nanoTime();
-            var response = config.getDurableExecutionClient()
-                    .getExecutionState(durableExecutionArn, checkpointToken, nextMarker);
-            logger.debug(
-                    "Durable getExecutionState API called (latency={}ns): {}.",
-                    System.nanoTime() - startTime,
-                    response);
-            operations.addAll(response.operations());
-            nextMarker = response.nextMarker();
+            try {
+                var response = config.getDurableExecutionClient()
+                        .getExecutionState(durableExecutionArn, checkpointToken, nextMarker);
+                logger.debug(
+                        "Durable getExecutionState API called (latency={}ns): {}.",
+                        System.nanoTime() - startTime,
+                        response);
+                operations.addAll(response.operations());
+                nextMarker = response.nextMarker();
+            } catch (AwsServiceException e) {
+                throw DurableApiErrorClassifier.classifyException(e);
+            }
         }
         return operations;
     }
@@ -187,35 +193,41 @@ private void checkpointBatch(List<OperationUpdate> updates) {
 
             var startTime = System.nanoTime();
             logger.debug("Calling durable checkpoint API with {} updates: {}", updates.size(), request);
-            var response = config.getDurableExecutionClient().checkpoint(durableExecutionArn, checkpointToken, request);
-            logger.debug("Durable checkpoint API called (latency={}ns): {}.", System.nanoTime() - startTime, response);
-
-            // Notify callback of completion
-            checkpointToken = response.checkpointToken();
-            if (response.newExecutionState() != null) {
-                // fetch all pages of operations
-                var operations = fetchAllPages(response.newExecutionState());
-
-                var processStartTime = System.nanoTime();
-                int completedFutures = 0;
+            try {
+                var response =
+                        config.getDurableExecutionClient().checkpoint(durableExecutionArn, checkpointToken, request);
                 logger.debug(
-                        "Processing {} operations. ({} pending pollers)", operations.size(), pollingFutures.size());
-                // call the callback
-                callback.accept(operations);
-
-                // complete the registered pollingFutures
-                for (var operation : operations) {
-                    var pollers = pollingFutures.remove(operation.id());
-                    if (pollers != null) {
-                        completedFutures += pollers.size();
-                        pollers.forEach(poller -> poller.complete(operation));
+                        "Durable checkpoint API called (latency={}ns): {}.", System.nanoTime() - startTime, response);
+
+                // Notify callback of completion
+                checkpointToken = response.checkpointToken();
+                if (response.newExecutionState() != null) {
+                    // fetch all pages of operations
+                    var operations = fetchAllPages(response.newExecutionState());
+
+                    var processStartTime = System.nanoTime();
+                    int completedFutures = 0;
+                    logger.debug(
+                            "Processing {} operations. ({} pending pollers)", operations.size(), pollingFutures.size());
+                    // call the callback
+                    callback.accept(operations);
+
+                    // complete the registered pollingFutures
+                    for (var operation : operations) {
+                        var pollers = pollingFutures.remove(operation.id());
+                        if (pollers != null) {
+                            completedFutures += pollers.size();
+                            pollers.forEach(poller -> poller.complete(operation));
+                        }
                     }
+                    logger.debug(
+                            "{} operations processed and {} pollers completed (latency={}ns). ",
+                            operations.size(),
+                            completedFutures,
+                            System.nanoTime() - processStartTime);
                 }
-                logger.debug(
-                        "{} operations processed and {} pollers completed (latency={}ns). ",
-                        operations.size(),
-                        completedFutures,
-                        System.nanoTime() - processStartTime);
+            } catch (AwsServiceException e) {
+                throw DurableApiErrorClassifier.classifyException(e);
             }
         }
     }

sdk/src/test/java/software/amazon/lambda/durable/exception/DurableApiErrorClassifierTest.java

@@ -0,0 +1,133 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+package software.amazon.lambda.durable.exception;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import java.util.stream.Stream;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import software.amazon.awssdk.awscore.exception.AwsErrorDetails;
+import software.amazon.awssdk.awscore.exception.AwsServiceException;
+import software.amazon.awssdk.http.SdkHttpResponse;
+
+class DurableApiErrorClassifierTest {
+
+    @ParameterizedTest
+    @MethodSource("kmsExceptions")
+    void classifyException_kmsError_returnsUnrecoverableDurableExecutionException(String errorCode, String message) {
+        var error = awsError(502, errorCode, message);
+        var result = DurableApiErrorClassifier.classifyException(error);
+        assertInstanceOf(UnrecoverableDurableExecutionException.class, result);
+        var unrecoverable = (UnrecoverableDurableExecutionException) result;
+        assertEquals(errorCode, unrecoverable.getErrorObject().errorType());
+        assertTrue(unrecoverable.getErrorObject().errorMessage().contains(message));
+    }
+
+    static Stream<Arguments> kmsExceptions() {
+        return Stream.of(
+                Arguments.of(
+                        "KMSAccessDeniedException",
+                        "Lambda was unable to decrypt the environment variables because KMS access was denied."
+                                + " Please check the function's KMS key settings."
+                                + " KMS Exception: AccessDeniedException"),
+                Arguments.of(
+                        "KMSDisabledException",
+                        "Lambda was unable to decrypt the environment variables because the KMS key used is disabled."
+                                + " Please check the function's KMS key settings."),
+                Arguments.of(
+                        "KMSInvalidStateException",
+                        "Lambda was unable to decrypt the environment variables because the KMS key is in an invalid"
+                                + " state for Decrypt. Please check the function's KMS key settings."),
+                Arguments.of(
+                        "KMSNotFoundException",
+                        "Lambda was unable to decrypt the environment variables because the KMS key was not found."
+                                + " Please check the function's KMS key settings."));
+    }
+
+    @Test
+    void classifyException_clientError4xx_returnsUnrecoverableDurableExecutionException() {
+        var error = awsError(403, "AccessDeniedException", "User is not authorized");
+
+        var result = DurableApiErrorClassifier.classifyException(error);
+        assertInstanceOf(UnrecoverableDurableExecutionException.class, result);
+    }
+
+    @Test
+    void classifyException_invalidCheckpointToken_returnsOriginalException() {
+        var error = awsError(400, "InvalidParameterValueException", "Invalid Checkpoint Token: token expired");
+
+        var result = DurableApiErrorClassifier.classifyException(error);
+        assertSame(error, result);
+    }
+
+    @Test
+    void classifyException_invalidParameterValueNonToken_returnsUnrecoverableDurableExecutionException() {
+        var error = awsError(
+                400,
+                "InvalidParameterValueException",
+                "The runtime parameter of python3.8 is no longer" + " supported for creating or updating functions.");
+
+        var result = DurableApiErrorClassifier.classifyException(error);
+        assertInstanceOf(UnrecoverableDurableExecutionException.class, result);
+    }
+
+    @Test
+    void classifyException_throttled429_returnsOriginalException() {
+        var error = awsError(429, "TooManyRequestsException", "Rate exceeded");
+
+        var result = DurableApiErrorClassifier.classifyException(error);
+        assertSame(error, result);
+    }
+
+    @Test
+    void classifyException_serverError500_returnsOriginalException() {
+        var error = awsError(500, "ServiceException", "Service encountered an error");
+
+        var result = DurableApiErrorClassifier.classifyException(error);
+        assertSame(error, result);
+    }
+
+    @Test
+    void classifyException_nonMatchingErrorCode502_returnsOriginalException() {
+        var error = awsError(502, "ServiceException", "Service unavailable");
+
+        var result = DurableApiErrorClassifier.classifyException(error);
+        assertSame(error, result);
+    }
+
+    @Test
+    void classifyException_errorDetailsPreserved() {
+        var error = awsError(
+                502,
+                "KMSAccessDeniedException",
+                "Lambda was unable to decrypt the environment variables because KMS access was denied."
+                        + " Please check the function's KMS key settings."
+                        + " KMS Exception: AccessDeniedException");
+
+        var result = DurableApiErrorClassifier.classifyException(error);
+        assertInstanceOf(UnrecoverableDurableExecutionException.class, result);
+
+        var unrecoverable = (UnrecoverableDurableExecutionException) result;
+        assertEquals("KMSAccessDeniedException", unrecoverable.getErrorObject().errorType());
+        assertTrue(unrecoverable
+                .getErrorObject()
+                .errorMessage()
+                .contains("Lambda was unable to decrypt the environment variables"));
+    }
+
+    private static AwsServiceException awsError(int statusCode, String errorCode, String message) {
+        return AwsServiceException.builder()
+                .message(message)
+                .awsErrorDetails(AwsErrorDetails.builder()
+                        .errorCode(errorCode)
+                        .errorMessage(message)
+                        .sdkHttpResponse(
+                                SdkHttpResponse.builder().statusCode(statusCode).build())
+                        .build())
+                .statusCode(statusCode)
+                .build();
+    }
+}