Skip to content

feat: [Parallel] Do not checkpoint parallel during replay #233

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
wangyb-A merged 3 commits into from
Mar 19, 2026
Merged

feat: [Parallel] Do not checkpoint parallel during replay #233

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
599b252
Do not checkpoint when replay
Mar 19, 2026
c274350
Update checkpoint condition
Mar 19, 2026
837c674
Remove runJoin(), add mock backend to fix tests
Mar 19, 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
2 changes: 2 additions & 0 deletions ...c/main/java/software/amazon/lambda/durable/examples/parallel/ParallelWithWaitExample.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,8 @@ public Output handleRequest(Input input, DurableContext context) {

var deliveries = futures.stream().map(DurableFuture::get).toList();
logger.info("All {} notifications delivered", deliveries.size());
// Test replay
context.wait("wait for finalization", Duration.ofSeconds(5));
return new Output(deliveries);
}
}
11 changes: 9 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 @@ -42,6 +42,8 @@
*/
public class ParallelOperation<T> extends ConcurrencyOperation<T> {

private boolean replaying = false;

public ParallelOperation(
OperationIdentifier operationIdentifier,
TypeToken<T> resultTypeToken,
Expand Down Expand Up @@ -79,6 +81,10 @@ protected <R> ChildContextOperation<R> createItem(

@Override
protected void handleSuccess() {
if (replaying) {
// Do not send checkpoint during replay
return;
}
sendOperationUpdate(OperationUpdate.builder()
.action(OperationAction.SUCCEED)
.subType(getSubType().getValue())
Expand All @@ -99,8 +105,9 @@ protected void start() {

@Override
protected void replay(Operation existing) {
// Always replay child branches for parallel
start();
// No-op: child branches handle their own replay via ChildContextOperation.replay().
// Set replaying=true so handleSuccess() skips re-checkpointing the already-completed parallel context.
replaying = true;
}

@Override
Expand Down
94 changes: 94 additions & 0 deletions sdk/src/test/java/software/amazon/lambda/durable/operation/ParallelOperationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -199,6 +199,100 @@ void contextHierarchy_branchesUseParallelContextAsParent() throws Exception {
assertNotNull(childOp);
}

// ===== Replay =====

@Test
void replay_doesNotSendStartCheckpoint() throws Exception {
// Simulate the parallel operation already existing in the service (STARTED status)
when(executionManager.getOperationAndUpdateReplayState(OPERATION_ID))
.thenReturn(Operation.builder()
.id(OPERATION_ID)
.name("test-parallel")
.type(OperationType.CONTEXT)
.subType(OperationSubType.PARALLEL.getValue())
.status(OperationStatus.STARTED)
.build());
// Both branches already succeeded
when(executionManager.getOperationAndUpdateReplayState("child-1"))
.thenReturn(Operation.builder()
.id("child-1")
.name("branch-1")
.type(OperationType.CONTEXT)
.subType(OperationSubType.PARALLEL_BRANCH.getValue())
.status(OperationStatus.SUCCEEDED)
.contextDetails(
ContextDetails.builder().result("\"r1\"").build())
.build());
when(executionManager.getOperationAndUpdateReplayState("child-2"))
.thenReturn(Operation.builder()
.id("child-2")
.name("branch-2")
.type(OperationType.CONTEXT)
.subType(OperationSubType.PARALLEL_BRANCH.getValue())
.status(OperationStatus.SUCCEEDED)
.contextDetails(
ContextDetails.builder().result("\"r2\"").build())
.build());

var op = createOperation(-1, -1, 0);
setOperationIdGenerator(op, mockIdGenerator);
op.execute();
op.addItem("branch-1", ctx -> "r1", TypeToken.get(String.class), SER_DES);
op.addItem("branch-2", ctx -> "r2", TypeToken.get(String.class), SER_DES);

runJoin(op);

verify(executionManager, never())
.sendOperationUpdate(argThat(update -> update.action() == OperationAction.START));
verify(executionManager, never())
.sendOperationUpdate(argThat(update -> update.action() == OperationAction.SUCCEED));
}

@Test
void replay_doesNotSendSucceedCheckpointWhenParallelAlreadySucceeded() throws Exception {
when(executionManager.getOperationAndUpdateReplayState(OPERATION_ID))
.thenReturn(Operation.builder()
.id(OPERATION_ID)
.name("test-parallel")
.type(OperationType.CONTEXT)
.subType(OperationSubType.PARALLEL.getValue())
.status(OperationStatus.SUCCEEDED)
.build());
when(executionManager.getOperationAndUpdateReplayState("child-1"))
.thenReturn(Operation.builder()
.id("child-1")
.name("branch-1")
.type(OperationType.CONTEXT)
.subType(OperationSubType.PARALLEL_BRANCH.getValue())
.status(OperationStatus.SUCCEEDED)
.contextDetails(
ContextDetails.builder().result("\"r1\"").build())
.build());
when(executionManager.getOperationAndUpdateReplayState("child-2"))
.thenReturn(Operation.builder()
.id("child-2")
.name("branch-2")
.type(OperationType.CONTEXT)
.subType(OperationSubType.PARALLEL_BRANCH.getValue())
.status(OperationStatus.SUCCEEDED)
.contextDetails(
ContextDetails.builder().result("\"r2\"").build())
.build());

var op = createOperation(-1, -1, 0);
setOperationIdGenerator(op, mockIdGenerator);
op.execute();
op.addItem("branch-1", ctx -> "r1", TypeToken.get(String.class), SER_DES);
op.addItem("branch-2", ctx -> "r2", TypeToken.get(String.class), SER_DES);

runJoin(op);

verify(executionManager, never())
.sendOperationUpdate(argThat(update -> update.action() == OperationAction.START));
verify(executionManager, never())
.sendOperationUpdate(argThat(update -> update.action() == OperationAction.SUCCEED));
}

// ===== handleFailure still sends SUCCEED =====

@Test
Expand Down
Loading