Skip to content

Commit f767128

Browse files
maschnetworkmaschnetwork
committed
Added testing for error scenarios
1 parent 738e7ed commit f767128

6 files changed

Lines changed: 294 additions & 2 deletions

File tree

‎examples/README.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -81,6 +81,7 @@ mvn test -Dtest=CloudBasedIntegrationTest \
8181
| [SimpleStepExample](src/main/java/com/amazonaws/lambda/durable/examples/SimpleStepExample.java) | Basic sequential steps |
8282
| [WaitExample](src/main/java/com/amazonaws/lambda/durable/examples/WaitExample.java) | Suspend execution with `wait()` |
8383
| [RetryExample](src/main/java/com/amazonaws/lambda/durable/examples/RetryExample.java) | Configuring retry strategies |
84+
| [ErrorHandlingExample](src/main/java/com/amazonaws/lambda/durable/examples/ErrorHandlingExample.java) | Handling `StepFailedException` and `StepInterruptedException` |
8485
| [GenericTypesExample](src/main/java/com/amazonaws/lambda/durable/examples/GenericTypesExample.java) | Working with `List<T>` and `Map<K,V>` |
8586
| [CustomConfigExample](src/main/java/com/amazonaws/lambda/durable/examples/CustomConfigExample.java) | Custom Lambda client and SerDes |
8687
| [WaitAtLeastExample](src/main/java/com/amazonaws/lambda/durable/examples/WaitAtLeastExample.java) | Concurrent `stepAsync()` with `wait()` |

‎examples/src/main/java/com/amazonaws/lambda/durable/examples/ErrorHandlingExample.java‎

Lines changed: 79 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,79 @@
1+
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
2+
// SPDX-License-Identifier: Apache-2.0
3+
package com.amazonaws.lambda.durable.examples;
4+
5+
import com.amazonaws.lambda.durable.DurableContext;
6+
import com.amazonaws.lambda.durable.DurableHandler;
7+
import com.amazonaws.lambda.durable.StepConfig;
8+
import com.amazonaws.lambda.durable.StepSemantics;
9+
import com.amazonaws.lambda.durable.exception.StepFailedException;
10+
import com.amazonaws.lambda.durable.exception.StepInterruptedException;
11+
import com.amazonaws.lambda.durable.retry.RetryStrategies;
12+
import org.slf4j.Logger;
13+
import org.slf4j.LoggerFactory;
14+
15+
/**
16+
* Example demonstrating error handling patterns with the Durable Execution SDK.
17+
*
18+
* <p>This example shows how to handle:
19+
* <ul>
20+
* <li>{@link StepFailedException} - when a step exhausts all retry attempts
21+
* <li>{@link StepInterruptedException} - when an AT_MOST_ONCE step is interrupted
22+
* </ul>
23+
*
24+
* <p>Note: {@code NonDeterministicExecutionException} is thrown by the SDK when code changes
25+
* between executions (e.g., step order/names changed). It should be fixed in code, not caught.
26+
*/
27+
public class ErrorHandlingExample extends DurableHandler<String, String> {
28+
29+
private static final Logger logger = LoggerFactory.getLogger(ErrorHandlingExample.class);
30+
31+
@Override
32+
public String handleRequest(String input, DurableContext context) {
33+
// Example 1: Catching StepFailedException with fallback logic
34+
String primaryResult;
35+
try {
36+
primaryResult = context.step(
37+
"call-primary-service",
38+
String.class,
39+
() -> {
40+
throw new RuntimeException("Primary service unavailable");
41+
},
42+
StepConfig.builder()
43+
.retryStrategy(RetryStrategies.Presets.NO_RETRY)
44+
.build());
45+
} catch (StepFailedException e) {
46+
logger.warn("Primary service failed, using fallback: {}", e.getMessage());
47+
primaryResult = context.step(
48+
"call-fallback-service",
49+
String.class,
50+
() -> "fallback-result");
51+
}
52+
53+
// Example 2: Handling StepInterruptedException for AT_MOST_ONCE operations
54+
// StepInterruptedException is thrown when an AT_MOST_ONCE step was started
55+
// but the function was interrupted before the step completed.
56+
// In normal execution, this step succeeds. The catch block handles the
57+
// interruption scenario that occurs during replay after an unexpected termination.
58+
String paymentResult;
59+
try {
60+
paymentResult = context.step(
61+
"charge-payment",
62+
String.class,
63+
() -> "payment-" + input,
64+
StepConfig.builder()
65+
.semantics(StepSemantics.AT_MOST_ONCE_PER_RETRY)
66+
.build());
67+
} catch (StepInterruptedException e) {
68+
logger.warn("Payment step interrupted, checking external status: {}", e.getOperationId());
69+
// In real code: check payment provider for transaction status
70+
// If payment went through, return success; otherwise, handle appropriately
71+
paymentResult = context.step(
72+
"verify-payment-status",
73+
String.class,
74+
() -> "verified-payment");
75+
}
76+
77+
return "Completed: " + primaryResult + ", " + paymentResult;
78+
}
79+
}

‎examples/src/test/java/com/amazonaws/lambda/durable/examples/ErrorHandlingExampleTest.java‎

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,54 @@
1+
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
2+
// SPDX-License-Identifier: Apache-2.0
3+
package com.amazonaws.lambda.durable.examples;
4+
5+
import static org.junit.jupiter.api.Assertions.*;
6+
7+
import com.amazonaws.lambda.durable.model.ExecutionStatus;
8+
import com.amazonaws.lambda.durable.testing.LocalDurableTestRunner;
9+
import org.junit.jupiter.api.Test;
10+
11+
class ErrorHandlingExampleTest {
12+
13+
@Test
14+
void testErrorHandlingWithFallback() {
15+
var handler = new ErrorHandlingExample();
16+
var runner = LocalDurableTestRunner.create(String.class, handler);
17+
18+
var result = runner.run("test-input");
19+
20+
assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
21+
assertTrue(result.getResult(String.class).contains("fallback-result"));
22+
}
23+
24+
@Test
25+
void testPaymentStepCompletes() {
26+
var handler = new ErrorHandlingExample();
27+
var runner = LocalDurableTestRunner.create(String.class, handler);
28+
29+
var result = runner.run("order-123");
30+
31+
assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
32+
// Normal execution: payment step succeeds with "payment-order-123"
33+
assertTrue(result.getResult(String.class).contains("payment-order-123"));
34+
}
35+
36+
@Test
37+
void testPaymentStepInterruptedRecovery() {
38+
var handler = new ErrorHandlingExample();
39+
var runner = LocalDurableTestRunner.create(String.class, handler);
40+
41+
// First run: both steps complete normally
42+
var result1 = runner.run("order-456");
43+
assertEquals(ExecutionStatus.SUCCEEDED, result1.getStatus());
44+
45+
// Simulate interruption: reset payment step to STARTED state
46+
runner.resetCheckpointToStarted("charge-payment");
47+
48+
// Second run: StepInterruptedException is caught, recovery step executes
49+
var result2 = runner.run("order-456");
50+
51+
assertEquals(ExecutionStatus.SUCCEEDED, result2.getStatus());
52+
assertTrue(result2.getResult(String.class).contains("verified-payment"));
53+
}
54+
}

‎sdk-integration-tests/src/test/java/com/amazonaws/lambda/durable/ExceptionIntegrationTest.java‎

Lines changed: 149 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,149 @@
1+
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
2+
// SPDX-License-Identifier: Apache-2.0
3+
package com.amazonaws.lambda.durable;
4+
5+
import static org.junit.jupiter.api.Assertions.*;
6+
7+
import com.amazonaws.lambda.durable.exception.StepFailedException;
8+
import com.amazonaws.lambda.durable.exception.StepInterruptedException;
9+
import com.amazonaws.lambda.durable.model.ExecutionStatus;
10+
import com.amazonaws.lambda.durable.retry.RetryStrategies;
11+
import com.amazonaws.lambda.durable.testing.LocalDurableTestRunner;
12+
import java.util.concurrent.atomic.AtomicInteger;
13+
import org.junit.jupiter.api.Test;
14+
15+
/**
16+
* Integration tests for exception handling scenarios documented in the README.
17+
*/
18+
class ExceptionIntegrationTest {
19+
20+
@Test
21+
void testStepFailedExceptionThrownAfterRetryExhaustion() {
22+
var runner = LocalDurableTestRunner.create(String.class, (input, ctx) -> {
23+
return ctx.step(
24+
"always-fails",
25+
String.class,
26+
() -> {
27+
throw new RuntimeException("Service unavailable");
28+
},
29+
StepConfig.builder()
30+
.retryStrategy(RetryStrategies.Presets.NO_RETRY)
31+
.build());
32+
});
33+
34+
var result = runner.run("test");
35+
36+
assertEquals(ExecutionStatus.FAILED, result.getStatus());
37+
}
38+
39+
@Test
40+
void testStepFailedExceptionCanBeCaughtWithFallback() {
41+
var runner = LocalDurableTestRunner.create(String.class, (input, ctx) -> {
42+
try {
43+
return ctx.step(
44+
"primary",
45+
String.class,
46+
() -> {
47+
throw new RuntimeException("Primary failed");
48+
},
49+
StepConfig.builder()
50+
.retryStrategy(RetryStrategies.Presets.NO_RETRY)
51+
.build());
52+
} catch (StepFailedException e) {
53+
return ctx.step("fallback", String.class, () -> "fallback-result");
54+
}
55+
});
56+
57+
var result = runner.run("test");
58+
59+
assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
60+
assertEquals("fallback-result", result.getResult(String.class));
61+
}
62+
63+
@Test
64+
void testStepInterruptedExceptionForAtMostOnceAfterCheckpointLoss() {
65+
var executionCount = new AtomicInteger(0);
66+
67+
var runner = LocalDurableTestRunner.create(String.class, (input, ctx) -> {
68+
return ctx.step(
69+
"at-most-once-step",
70+
String.class,
71+
() -> {
72+
executionCount.incrementAndGet();
73+
return "result";
74+
},
75+
StepConfig.builder()
76+
.semantics(StepSemantics.AT_MOST_ONCE_PER_RETRY)
77+
.build());
78+
});
79+
80+
// First run succeeds
81+
runner.run("test");
82+
assertEquals(1, executionCount.get());
83+
84+
// Simulate checkpoint loss (step started but result not saved)
85+
runner.resetCheckpointToStarted("at-most-once-step");
86+
87+
// Second run should fail with StepInterruptedException (not re-execute)
88+
var result = runner.run("test");
89+
90+
assertEquals(ExecutionStatus.FAILED, result.getStatus());
91+
assertEquals(1, executionCount.get()); // Should NOT have re-executed
92+
}
93+
94+
@Test
95+
void testStepInterruptedExceptionCanBeCaughtForRecovery() {
96+
var executionCount = new AtomicInteger(0);
97+
98+
var runner = LocalDurableTestRunner.create(String.class, (input, ctx) -> {
99+
try {
100+
return ctx.step(
101+
"payment",
102+
String.class,
103+
() -> {
104+
executionCount.incrementAndGet();
105+
return "payment-success";
106+
},
107+
StepConfig.builder()
108+
.semantics(StepSemantics.AT_MOST_ONCE_PER_RETRY)
109+
.build());
110+
} catch (StepInterruptedException e) {
111+
// Recovery: check external status and return verified result
112+
return ctx.step("verify-payment", String.class, () -> "verified-payment");
113+
}
114+
});
115+
116+
// First run succeeds
117+
runner.run("test");
118+
119+
// Simulate interruption (step started but result not checkpointed)
120+
runner.resetCheckpointToStarted("payment");
121+
122+
// Second run catches exception and recovers
123+
var result = runner.run("test");
124+
125+
assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
126+
assertEquals("verified-payment", result.getResult(String.class));
127+
}
128+
129+
@Test
130+
void testNonDeterministicExceptionOnStepNameChange() {
131+
var useNewName = new AtomicInteger(0);
132+
133+
var runner = LocalDurableTestRunner.create(String.class, (input, ctx) -> {
134+
var stepName = useNewName.get() == 0 ? "original-name" : "changed-name";
135+
return ctx.step(stepName, String.class, () -> "result");
136+
});
137+
138+
// First run with original name
139+
runner.run("test");
140+
141+
// Change step name for replay
142+
useNewName.set(1);
143+
144+
// Replay should detect non-determinism
145+
var result = runner.run("test");
146+
147+
assertEquals(ExecutionStatus.FAILED, result.getStatus());
148+
}
149+
}

‎sdk-testing/src/main/java/com/amazonaws/lambda/durable/testing/LocalMemoryExecutionClient.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -168,7 +168,7 @@ private StepDetails buildStepDetails(OperationUpdate update) {
168168

169169
var detailsBuilder = existing != null ? existing.toBuilder() : StepDetails.builder();
170170

171-
if (update.action() == OperationAction.RETRY) {
171+
if (update.action() == OperationAction.RETRY || update.action() == OperationAction.FAIL) {
172172
var attempt = existing != null && existing.attempt() != null ? existing.attempt() + 1 : 1;
173173
detailsBuilder.attempt(attempt).error(update.error());
174174
}

‎sdk/src/main/java/com/amazonaws/lambda/durable/operation/StepOperation.java‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -334,10 +334,19 @@ public T get() {
334334
// SneakyThrow.sneakyThrow((Throwable)
335335
// serDes.deserializeWithTypeInfo(errorData));
336336
// }
337+
338+
var errorType = op.stepDetails().error().errorType();
339+
340+
// Throw StepInterruptedException directly for AT_MOST_ONCE interrupted steps
341+
// Todo: Change once errorData object is implemented
342+
if ("StepInterruptedException".equals(errorType)) {
343+
throw new StepInterruptedException(operationId, name);
344+
}
345+
337346
throw new StepFailedException(
338347
String.format(
339348
"Step failed with error of type %s. Message: %s",
340-
op.stepDetails().error().errorType(),
349+
errorType,
341350
op.stepDetails().error().errorMessage()),
342351
null,
343352
// Preserve original stack trace

0 commit comments

Comments
 (0)