diff --git a/sdk/src/main/java/software/amazon/lambda/durable/execution/ExecutionManager.java b/sdk/src/main/java/software/amazon/lambda/durable/execution/ExecutionManager.java index 007df8665..1743aefaa 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/execution/ExecutionManager.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/execution/ExecutionManager.java @@ -260,17 +260,13 @@ void onCheckpointComplete(List newOperations) { }); }); - // Fire onOperationChange when a checkpoint response changed one or more operations - if (!updatedOperations.isEmpty()) { + // Fire onOperationChange when a checkpoint response changed one or more operations. Only plugins read + // the change info, and building it converts every stored operation. So skip it when no plugin is registered. + var pluginRunner = durableConfig.getPluginRunner(); + if (!updatedOperations.isEmpty() && !pluginRunner.isEmpty()) { var requestId = lambdaContext != null ? lambdaContext.getAwsRequestId() : null; - durableConfig - .getPluginRunner() - .onOperationChange(PluginInfoConverter.toOperationChangeInfo( - requestId, - durableExecutionArn, - updatedOperations, - operationStorage.values(), - initialOperationIds)); + pluginRunner.onOperationChange(PluginInfoConverter.toOperationChangeInfo( + requestId, durableExecutionArn, updatedOperations, operationStorage.values(), initialOperationIds)); } } diff --git a/sdk/src/test/java/software/amazon/lambda/durable/execution/ExecutionManagerTest.java b/sdk/src/test/java/software/amazon/lambda/durable/execution/ExecutionManagerTest.java index 7bb872911..52e93353f 100644 --- a/sdk/src/test/java/software/amazon/lambda/durable/execution/ExecutionManagerTest.java +++ b/sdk/src/test/java/software/amazon/lambda/durable/execution/ExecutionManagerTest.java @@ -4,7 +4,10 @@ import static org.junit.jupiter.api.Assertions.*; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.CALLS_REAL_METHODS; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockStatic; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.when; import java.time.Instant; @@ -12,6 +15,7 @@ import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; import java.util.concurrent.ConcurrentHashMap; @@ -24,6 +28,7 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; +import java.util.stream.Collectors; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.CsvSource; @@ -46,6 +51,9 @@ import software.amazon.lambda.durable.model.OperationIdentifier; import software.amazon.lambda.durable.model.OperationSubType; import software.amazon.lambda.durable.operation.BaseDurableOperation; +import software.amazon.lambda.durable.plugin.DurableExecutionPlugin; +import software.amazon.lambda.durable.plugin.OperationChangeInfo; +import software.amazon.lambda.durable.plugin.PluginInfoConverter; class ExecutionManagerTest { private static final String EXECUTION_OP_ID = "01234567-0123-0123-0123-012345678901"; @@ -55,13 +63,16 @@ class ExecutionManagerTest { private DurableExecutionClient client; - private ExecutionManager createManager(List operations) { + private ExecutionManager createManager(List operations, DurableExecutionPlugin... plugins) { client = TestUtils.createMockClient(); var initialState = CheckpointUpdatedExecutionState.builder().operations(operations).build(); return new ExecutionManager( new DurableExecutionInput(EXECUTION_ARN, "test-token", initialState), - DurableConfig.builder().withDurableExecutionClient(client).build(), + DurableConfig.builder() + .withDurableExecutionClient(client) + .withPlugins(plugins) + .build(), null); } @@ -506,6 +517,79 @@ protected void replay(Operation existing) {} manager.getOperationAndUpdateReplayState("step").status()); } + @Test + void checkpointWithoutPluginsSkipsOperationChangeInfo() { + try (var manager = createManager(List.of(executionOp(), stepOp("step", OperationStatus.PENDING)))) { + var operation = registerStepOperation(manager); + try (var converter = mockStatic(PluginInfoConverter.class, CALLS_REAL_METHODS)) { + manager.onCheckpointComplete( + List.of(stepOp("step", OperationStatus.SUCCEEDED), stepOp("other", OperationStatus.SUCCEEDED))); + + converter.verify( + () -> PluginInfoConverter.toOperationChangeInfo(any(), any(), any(), any(), any()), never()); + } + + // The checkpoint still stores every operation and completes the waiting operation. + assertTrue(operation.getCompletionFuture().isDone()); + assertEquals( + Map.of( + EXECUTION_OP_ID, + OperationStatus.STARTED, + "step", + OperationStatus.SUCCEEDED, + "other", + OperationStatus.SUCCEEDED), + statusesById(manager)); + } + } + + @Test + void checkpointWithPluginReportsEachStatusChangeOnce() { + var changes = new ArrayList(); + var plugin = new DurableExecutionPlugin() { + @Override + public void onOperationChange(OperationChangeInfo info) { + changes.add(info); + } + }; + try (var manager = createManager(List.of(executionOp(), stepOp("step", OperationStatus.PENDING)), plugin)) { + var response = List.of(stepOp("step", OperationStatus.SUCCEEDED), stepOp("other", OperationStatus.STARTED)); + manager.onCheckpointComplete(response); + // The same response again changes no status, so it reports no change. + manager.onCheckpointComplete(response); + + assertEquals(1, changes.size()); + var change = changes.get(0); + assertEquals(Set.of("step", "other"), change.updatedOperations().keySet()); + assertEquals( + OperationStatus.SUCCEEDED, + change.updatedOperations().get("step").status()); + assertEquals( + Set.of(EXECUTION_OP_ID, "step", "other"), + change.operations().keySet()); + assertEquals( + OperationStatus.STARTED, change.operations().get("other").status()); + } + } + + /** Registers a step operation with the manager, so a checkpoint that reports "step" notifies it. */ + private BaseDurableOperation registerStepOperation(ExecutionManager manager) { + var durableContext = mock(DurableContextImpl.class); + when(durableContext.getExecutionManager()).thenReturn(manager); + return new BaseDurableOperation( + OperationIdentifier.of("step", "step", OperationSubType.STEP), durableContext, null) { + @Override + protected void start() {} + + @Override + protected void replay(Operation existing) {} + }; + } + + private static Map statusesById(ExecutionManager manager) { + return manager.getOperationsSnapshot().stream().collect(Collectors.toMap(Operation::id, Operation::status)); + } + @Test void operationProcessCheckpointUpdateDelegatesThroughManagerLock() throws Exception { var manager = createManager(List.of(executionOp(), stepOp("step", OperationStatus.PENDING)));