Skip to content

fix: [Parallel] fix NPE in handle complete #279

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
zhongkechen merged 3 commits into from
Mar 27, 2026
+98 −3
Merged

fix: [Parallel] fix NPE in handle complete #279

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
65a7975
Add basic parallel integration tests
Mar 25, 2026
42e1e0e
Fix parallel NPE in handle complete
Mar 26, 2026
f852e87
Merge branch 'main' into parallel/npe
wangyb-A Mar 26, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
95 changes: 94 additions & 1 deletion ...tegration-tests/src/test/java/software/amazon/lambda/durable/ParallelIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -227,7 +227,7 @@ void testParallelReplayAfterInterruption_cachedResultsUsed() {
var firstRunCount = executionCounts.get();
assertTrue(firstRunCount >= 3, "Expected at least 3 executions on first run but got " + firstRunCount);

var result2 = runner.run("test");
var result2 = runner.runUntilComplete("test");
assertEquals(ExecutionStatus.SUCCEEDED, result2.getStatus());
assertEquals("A,B,C", result2.getResult(String.class));
assertEquals(firstRunCount, executionCounts.get(), "Branch functions should not re-execute on replay");
Expand Down Expand Up @@ -536,4 +536,97 @@ void testParallelResultSummary_succeededAndFailedCounts() {
assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
assertEquals("3/2", result.getResult(String.class));
}

@Test
void testParallelWithToleratedFailureCount_earlyTermination() {
var runner = LocalDurableTestRunner.create(String.class, (input, context) -> {
var config = ParallelConfig.builder()
.maxConcurrency(1)
.completionConfig(CompletionConfig.toleratedFailureCount(1))
.build();
var futures = new ArrayList<DurableFuture<String>>();
var parallel = context.parallel("tolerated-fail", config);

try (parallel) {
futures.add(parallel.branch("branch-ok", String.class, ctx -> "OK"));
futures.add(parallel.branch("branch-fail1", String.class, ctx -> {
throw new RuntimeException("failed: fail1");
}));
futures.add(parallel.branch("branch-fail2", String.class, ctx -> {
throw new RuntimeException("failed: fail2");
}));
futures.add(parallel.branch("branch-ok2", String.class, ctx -> "OK2"));
}

var result = parallel.get();
assertEquals(ConcurrencyCompletionStatus.FAILURE_TOLERANCE_EXCEEDED, result.completionStatus());
assertFalse(result.completionStatus().isSucceeded());
assertEquals(4, result.size());
assertEquals("OK", futures.get(0).get());

return "done";
});

var result = runner.runUntilComplete("test");
assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
}

@Test
void testParallelWithMinSuccessful_earlyTermination() {
var runner = LocalDurableTestRunner.create(String.class, (input, context) -> {
var config = ParallelConfig.builder()
.maxConcurrency(1)
.completionConfig(CompletionConfig.minSuccessful(2))
.build();
var futures = new ArrayList<DurableFuture<String>>();
var parallel = context.parallel("min-successful", config);

try (parallel) {
for (var item : List.of("a", "b", "c", "d", "e")) {
futures.add(parallel.branch("branch-" + item, String.class, ctx -> item.toUpperCase()));
}
}

var result = parallel.get();
assertEquals(ConcurrencyCompletionStatus.MIN_SUCCESSFUL_REACHED, result.completionStatus());
assertTrue(result.completionStatus().isSucceeded());
assertEquals(5, result.size());
assertEquals("A", futures.get(0).get());
assertEquals("B", futures.get(1).get());

return "done";
});

var result = runner.runUntilComplete("test");
assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
}

@Test
void testParallelWithAllSuccessful_stopsOnFirstFailure() {
var runner = LocalDurableTestRunner.create(String.class, (input, context) -> {
var config = ParallelConfig.builder()
.maxConcurrency(1)
.completionConfig(CompletionConfig.allSuccessful())
.build();
var futures = new ArrayList<DurableFuture<String>>();
var parallel = context.parallel("all-successful", config);

try (parallel) {
futures.add(parallel.branch("branch-ok1", String.class, ctx -> "OK1"));
futures.add(parallel.branch("branch-fail", String.class, ctx -> {
throw new RuntimeException("failed");
}));
futures.add(parallel.branch("branch-ok2", String.class, ctx -> "OK2"));
}

var result = parallel.get();
assertEquals(ConcurrencyCompletionStatus.FAILURE_TOLERANCE_EXCEEDED, result.completionStatus());
assertEquals("OK1", futures.get(0).get());

return "done";
});

var result = runner.runUntilComplete("test");
assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
}
}
6 changes: 4 additions & 2 deletions sdk/src/main/java/software/amazon/lambda/durable/operation/ParallelOperation.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,10 +68,12 @@ public ParallelOperation(
protected void handleCompletion(ConcurrencyCompletionStatus concurrencyCompletionStatus) {
var items = getBranches();
int succeededCount = Math.toIntExact(items.stream()
.filter(item -> item.getOperation().status() == OperationStatus.SUCCEEDED)
.filter(item ->
item.getOperation() != null && item.getOperation().status() == OperationStatus.SUCCEEDED)
.count());
int failedCount = Math.toIntExact(items.stream()
.filter(item -> item.getOperation().status() != OperationStatus.SUCCEEDED)
.filter(item ->
item.getOperation() != null && item.getOperation().status() != OperationStatus.SUCCEEDED)
.count());
this.cachedResult = new ParallelResult(items.size(), succeededCount, failedCount, concurrencyCompletionStatus);
if (skipCheckpoint) {
Expand Down
Loading