From 11a94fd00e51acf432812dcdf302bcaa1abdc4bc Mon Sep 17 00:00:00 2001 From: nvasiu Date: Fri, 10 Jul 2026 18:25:50 +0000 Subject: [PATCH] feat(plugin): add onOperationChange hook --- .../lambda/durable/PluginIntegrationTest.java | 69 +++++++++++++++++++ .../durable/execution/DurableExecutor.java | 2 +- .../durable/execution/ExecutionManager.java | 21 +++++- .../operation/BaseDurableOperation.java | 4 +- .../plugin/DurableExecutionPlugin.java | 7 ++ .../durable/plugin/OperationChangeInfo.java | 23 +++++++ .../plugin/OperationChangeItemInfo.java | 32 +++++++++ .../durable/plugin/PluginInfoConverter.java | 42 +++++++++++ .../lambda/durable/plugin/PluginRunner.java | 4 ++ .../lambda/durable/DurableContextTest.java | 3 +- .../lambda/durable/ReplayValidationTest.java | 3 +- .../durable/context/BaseContextImplTest.java | 3 +- .../context/DurableContextImplTest.java | 3 +- .../execution/ExecutionManagerTest.java | 15 ++-- .../operation/CallbackOperationTest.java | 3 +- .../durable/plugin/PluginRunnerTest.java | 13 ++++ 16 files changed, 233 insertions(+), 14 deletions(-) create mode 100644 sdk/src/main/java/software/amazon/lambda/durable/plugin/OperationChangeInfo.java create mode 100644 sdk/src/main/java/software/amazon/lambda/durable/plugin/OperationChangeItemInfo.java diff --git a/sdk-integration-tests/src/test/java/software/amazon/lambda/durable/PluginIntegrationTest.java b/sdk-integration-tests/src/test/java/software/amazon/lambda/durable/PluginIntegrationTest.java index 83d2b7b59..6b8cd2c10 100644 --- a/sdk-integration-tests/src/test/java/software/amazon/lambda/durable/PluginIntegrationTest.java +++ b/sdk-integration-tests/src/test/java/software/amazon/lambda/durable/PluginIntegrationTest.java @@ -11,6 +11,7 @@ import java.util.concurrent.atomic.AtomicInteger; import org.junit.jupiter.api.Test; import software.amazon.awssdk.services.lambda.model.ErrorObject; +import software.amazon.awssdk.services.lambda.model.OperationStatus; import software.amazon.lambda.durable.config.StepConfig; import software.amazon.lambda.durable.model.ExecutionStatus; import software.amazon.lambda.durable.model.WaitForConditionResult; @@ -330,6 +331,63 @@ void plugin_operationEnd_noError_whenOperationSucceeds() { assertNull(stepEnd.error(), "error should be null for successful step"); } + // ─── Operation change hook ─────────────────────────────────────────── + + @Test + void plugin_receivesOperationChange_forStep() { + var plugin = new RecordingPlugin(); + var config = DurableConfig.builder().withPlugins(plugin).build(); + + var runner = LocalDurableTestRunner.create( + String.class, (input, context) -> context.step("my-step", String.class, stepCtx -> "result"), config); + + runner.runUntilComplete("input"); + + // A checkpoint response that reports the step at a new status should fire onOperationChange + assertFalse( + plugin.operationChanges.isEmpty(), "onOperationChange should fire when an operation changes status"); + + var changeWithStep = plugin.operationChanges.stream() + .filter(info -> info.updatedOperations().values().stream().anyMatch(op -> "my-step".equals(op.name()))) + .findFirst() + .orElse(null); + assertNotNull(changeWithStep, "onOperationChange should include the 'my-step' operation that changed status"); + assertNotNull(changeWithStep.durableExecutionArn()); + assertFalse(changeWithStep.operations().isEmpty(), "snapshot of all operations should be present"); + } + + @Test + void plugin_operationChange_includesErrorAndStatus_whenStepFails() { + var plugin = new RecordingPlugin(); + var config = DurableConfig.builder().withPlugins(plugin).build(); + + var runner = LocalDurableTestRunner.create( + String.class, + (input, context) -> context.step( + "failing-step", + String.class, + stepCtx -> { + throw new RuntimeException("step exploded"); + }, + StepConfig.builder() + .retryStrategy(RetryStrategies.Presets.NO_RETRY) + .build()), + config); + + var result = runner.run("input"); + assertEquals(ExecutionStatus.FAILED, result.getStatus()); + + // The failing step transitions through several statuses; find the one reporting FAILED + var failedItem = plugin.operationChanges.stream() + .flatMap(info -> info.updatedOperations().values().stream()) + .filter(op -> "failing-step".equals(op.name()) && op.status() == OperationStatus.FAILED) + .findFirst() + .orElse(null); + assertNotNull(failedItem, "onOperationChange should report 'failing-step' with FAILED status"); + assertNotNull(failedItem.error(), "error should be propagated for the failed step"); + assertTrue(failedItem.error().getMessage().contains("step exploded")); + } + // ─── User function hooks ───────────────────────────────────────────── @Test @@ -522,6 +580,7 @@ private static class RecordingPlugin implements DurableExecutionPlugin { final List operationEnds = Collections.synchronizedList(new ArrayList<>()); final List userFunctionStarts = Collections.synchronizedList(new ArrayList<>()); final List userFunctionEnds = Collections.synchronizedList(new ArrayList<>()); + final List operationChanges = Collections.synchronizedList(new ArrayList<>()); @Override public void onInvocationStart(InvocationInfo info) { @@ -552,6 +611,11 @@ public void onUserFunctionStart(UserFunctionStartInfo info) { public void onUserFunctionEnd(UserFunctionEndInfo info) { userFunctionEnds.add(info); } + + @Override + public void onOperationChange(OperationChangeInfo info) { + operationChanges.add(info); + } } /** Plugin that throws on every hook to verify error isolation. */ @@ -585,5 +649,10 @@ public void onUserFunctionStart(UserFunctionStartInfo info) { public void onUserFunctionEnd(UserFunctionEndInfo info) { throw new RuntimeException("plugin error"); } + + @Override + public void onOperationChange(OperationChangeInfo info) { + throw new RuntimeException("plugin error"); + } } } diff --git a/sdk/src/main/java/software/amazon/lambda/durable/execution/DurableExecutor.java b/sdk/src/main/java/software/amazon/lambda/durable/execution/DurableExecutor.java index a8a90f6fb..2ecbfff50 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/execution/DurableExecutor.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/execution/DurableExecutor.java @@ -54,7 +54,7 @@ public static DurableExecutionOutput execute( BiFunction handler, DurableConfig config) { var pluginRunner = config.getPluginRunner(); - try (var executionManager = new ExecutionManager(input, config)) { + try (var executionManager = new ExecutionManager(input, config, lambdaContext)) { var isFirstInvocation = !executionManager.isReplaying(); var requestId = lambdaContext != null ? lambdaContext.getAwsRequestId() : null; var executionArn = input.durableExecutionArn(); 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 ef5c4e456..1c45cb0d6 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 @@ -2,6 +2,7 @@ // SPDX-License-Identifier: Apache-2.0 package software.amazon.lambda.durable.execution; +import com.amazonaws.services.lambda.runtime.Context; import java.time.Instant; import java.util.ArrayList; import java.util.Collections; @@ -27,6 +28,7 @@ import software.amazon.lambda.durable.model.DurableExecutionInput; import software.amazon.lambda.durable.model.SafeCloseable; import software.amazon.lambda.durable.operation.BaseDurableOperation; +import software.amazon.lambda.durable.plugin.PluginInfoConverter; /** * Central manager for durable execution coordination. @@ -57,6 +59,7 @@ public class ExecutionManager implements SafeCloseable { private final Map operationStorage; private final Operation executionOp; private final String durableExecutionArn; + private final Context lambdaContext; private final AtomicReference executionMode; private final DurableConfig durableConfig; private final Set updatedOperationIdsSinceLastInvocation; @@ -70,9 +73,10 @@ public class ExecutionManager implements SafeCloseable { // ===== Checkpoint Batching ===== private final CheckpointManager checkpointManager; - public ExecutionManager(DurableExecutionInput input, DurableConfig config) { + public ExecutionManager(DurableExecutionInput input, DurableConfig config, Context lambdaContext) { durableConfig = config; this.durableExecutionArn = input.durableExecutionArn(); + this.lambdaContext = lambdaContext; // Store the set of operation IDs updated since the last successful invocation this.updatedOperationIdsSinceLastInvocation = @@ -136,7 +140,13 @@ public void registerOperation(BaseDurableOperation operation) { // ===== Checkpoint Completion Handler ===== /** Called by CheckpointManager when a checkpoint completes. Updates operationStorage and notify operations . */ private void onCheckpointComplete(List newOperations) { + var updatedOperations = new ArrayList(); newOperations.forEach(op -> { + // Detect a status change against the previously stored operation + var previous = operationStorage.get(op.id()); + if (previous == null || previous.status() != op.status()) { + updatedOperations.add(op); + } // Update operation storage operationStorage.put(op.id(), op); // call registered operation's onCheckpointComplete method for completed operations @@ -145,6 +155,15 @@ private void onCheckpointComplete(List newOperations) { return operation; }); }); + + // Fire onOperationChange when a checkpoint response changed one or more operations + if (!updatedOperations.isEmpty()) { + var requestId = lambdaContext != null ? lambdaContext.getAwsRequestId() : null; + durableConfig + .getPluginRunner() + .onOperationChange(PluginInfoConverter.toOperationChangeInfo( + requestId, durableExecutionArn, updatedOperations, operationStorage.values())); + } } /** diff --git a/sdk/src/main/java/software/amazon/lambda/durable/operation/BaseDurableOperation.java b/sdk/src/main/java/software/amazon/lambda/durable/operation/BaseDurableOperation.java index 63a5d8bd6..1527241ea 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/operation/BaseDurableOperation.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/operation/BaseDurableOperation.java @@ -535,7 +535,7 @@ protected void fireOnOperationEnd(Operation operation, Throwable error, boolean * Extracts the error from a terminal operation as a Throwable. Returns null if the operation succeeded or has no * error details. */ - private Throwable extractErrorFromOperation(Operation operation) { + public static Throwable extractErrorFromOperation(Operation operation) { if (operation.status() != OperationStatus.FAILED && operation.status() != OperationStatus.TIMED_OUT && operation.status() != OperationStatus.STOPPED) { @@ -549,7 +549,7 @@ private Throwable extractErrorFromOperation(Operation operation) { } /** Extracts the ErrorObject from an operation based on its type. */ - private static ErrorObject getErrorObject(Operation operation) { + public static ErrorObject getErrorObject(Operation operation) { if (operation.type() == null) { return null; } diff --git a/sdk/src/main/java/software/amazon/lambda/durable/plugin/DurableExecutionPlugin.java b/sdk/src/main/java/software/amazon/lambda/durable/plugin/DurableExecutionPlugin.java index 7aee46640..5db30a4cd 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/plugin/DurableExecutionPlugin.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/plugin/DurableExecutionPlugin.java @@ -51,6 +51,13 @@ default void onOperationStart(OperationInfo info) {} */ default void onOperationEnd(OperationEndInfo info) {} + /** + * Called when a checkpoint response changes the status of one or more operations. + * + *

Receives the operations that changed and a snapshot of all operations. + */ + default void onOperationChange(OperationChangeInfo info) {} + // ─── User function hooks ───────────────────────────────────────────── /** diff --git a/sdk/src/main/java/software/amazon/lambda/durable/plugin/OperationChangeInfo.java b/sdk/src/main/java/software/amazon/lambda/durable/plugin/OperationChangeInfo.java new file mode 100644 index 000000000..a1d29fedd --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/plugin/OperationChangeInfo.java @@ -0,0 +1,23 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.plugin; + +import java.util.Map; + +/** + * Information provided when a checkpoint response changes one or more operations. + * + *

Holds a map with an {@link OperationChangeItemInfo} for every operation that changed in the checkpoint response. + * + * @param requestId the Lambda request ID for the invocation that observed the change + * @param durableExecutionArn the durable execution ARN + * @param updatedOperations operations whose status changed in this checkpoint response, keyed by operation ID + * @param operations a snapshot of all known operations after the update, keyed by operation ID + * @deprecated This is a preview API that is experimental and may be changed or removed in future releases. + */ +@Deprecated +public record OperationChangeInfo( + String requestId, + String durableExecutionArn, + Map updatedOperations, + Map operations) {} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/plugin/OperationChangeItemInfo.java b/sdk/src/main/java/software/amazon/lambda/durable/plugin/OperationChangeItemInfo.java new file mode 100644 index 000000000..6846280fe --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/plugin/OperationChangeItemInfo.java @@ -0,0 +1,32 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.plugin; + +import java.time.Instant; +import software.amazon.awssdk.services.lambda.model.OperationStatus; + +/** + * Operation-level information for a single operation within an {@link OperationChangeInfo}. + * + * @param id operation ID + * @param name human-readable operation name (may be null) + * @param type operation type + * @param subType operation sub-type (may be null) + * @param parentId parent operation ID (null for root-level operations) + * @param startTimestamp when the operation started + * @param endTimestamp when the operation ended + * @param error non-null if the operation failed + * @param status operation status + * @deprecated This is a preview API that is experimental and may be changed or removed in future releases. + */ +@Deprecated +public record OperationChangeItemInfo( + String id, + String name, + String type, + String subType, + String parentId, + Instant startTimestamp, + Instant endTimestamp, + Throwable error, + OperationStatus status) {} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/plugin/PluginInfoConverter.java b/sdk/src/main/java/software/amazon/lambda/durable/plugin/PluginInfoConverter.java index 87260151a..b5f869efe 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/plugin/PluginInfoConverter.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/plugin/PluginInfoConverter.java @@ -3,8 +3,11 @@ package software.amazon.lambda.durable.plugin; import java.time.Instant; +import java.util.Collection; +import java.util.stream.Collectors; import software.amazon.awssdk.services.lambda.model.Operation; import software.amazon.lambda.durable.model.OperationIdentifier; +import software.amazon.lambda.durable.operation.BaseDurableOperation; /** * Utility methods for converting SDK internal types to plugin info records. @@ -108,4 +111,43 @@ public static UserFunctionEndInfo toUserFunctionEndInfo( succeeded, error); } + + /** + * Creates an {@link OperationChangeInfo} from the durable operations whose status changed in a checkpoint response + * and a snapshot of all operations tracked for the execution. + * + * @param requestId the Lambda request ID for the invocation + * @param durableExecutionArn the durable execution ARN + * @param updatedOperations the durable operations whose status changed in this checkpoint response + * @param allOperations all durable operations tracked for the execution after this response + * @return an OperationChangeInfo record + */ + public static OperationChangeInfo toOperationChangeInfo( + String requestId, + String durableExecutionArn, + Collection updatedOperations, + Collection allOperations) { + return new OperationChangeInfo( + requestId, + durableExecutionArn, + updatedOperations.stream() + .collect(Collectors.toUnmodifiableMap( + Operation::id, PluginInfoConverter::toOperationChangeItemInfo)), + allOperations.stream() + .collect(Collectors.toUnmodifiableMap( + Operation::id, PluginInfoConverter::toOperationChangeItemInfo))); + } + + private static OperationChangeItemInfo toOperationChangeItemInfo(Operation operation) { + return new OperationChangeItemInfo( + operation.id(), + operation.name(), + operation.typeAsString(), + operation.subType(), + operation.parentId(), + operation.startTimestamp(), + operation.endTimestamp(), + BaseDurableOperation.extractErrorFromOperation(operation), + operation.status()); + } } diff --git a/sdk/src/main/java/software/amazon/lambda/durable/plugin/PluginRunner.java b/sdk/src/main/java/software/amazon/lambda/durable/plugin/PluginRunner.java index 4620fd062..78837a492 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/plugin/PluginRunner.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/plugin/PluginRunner.java @@ -78,6 +78,10 @@ public void onOperationEnd(OperationEndInfo info) { run(p -> p.onOperationEnd(info)); } + public void onOperationChange(OperationChangeInfo info) { + run(p -> p.onOperationChange(info)); + } + public void onUserFunctionStart(UserFunctionStartInfo info) { run(p -> p.onUserFunctionStart(info)); } diff --git a/sdk/src/test/java/software/amazon/lambda/durable/DurableContextTest.java b/sdk/src/test/java/software/amazon/lambda/durable/DurableContextTest.java index 1d50cc1c8..4b46c8301 100644 --- a/sdk/src/test/java/software/amazon/lambda/durable/DurableContextTest.java +++ b/sdk/src/test/java/software/amazon/lambda/durable/DurableContextTest.java @@ -46,7 +46,8 @@ private DurableContext createTestContext(List initialOperations) { CheckpointUpdatedExecutionState.builder().operations(operations).build(); var executionManager = new ExecutionManager( new DurableExecutionInput(EXECUTION_ARN, "test-token", initialExecutionState), - DurableConfig.builder().withDurableExecutionClient(client).build()); + DurableConfig.builder().withDurableExecutionClient(client).build(), + null); var root = DurableContextImpl.createRootContext( executionManager, software.amazon.lambda.durable.DurableConfig.builder().build(), diff --git a/sdk/src/test/java/software/amazon/lambda/durable/ReplayValidationTest.java b/sdk/src/test/java/software/amazon/lambda/durable/ReplayValidationTest.java index cdbe0711b..98f505ad6 100644 --- a/sdk/src/test/java/software/amazon/lambda/durable/ReplayValidationTest.java +++ b/sdk/src/test/java/software/amazon/lambda/durable/ReplayValidationTest.java @@ -44,7 +44,8 @@ private DurableContext createTestContext(List initialOperations) { CheckpointUpdatedExecutionState.builder().operations(operations).build(); var executionManager = new ExecutionManager( new DurableExecutionInput(EXECUTION_ARN, "test-token", initialExecutionState), - DurableConfig.builder().withDurableExecutionClient(client).build()); + DurableConfig.builder().withDurableExecutionClient(client).build(), + null); var context = DurableContextImpl.createRootContext( executionManager, DurableConfig.builder().build(), null); executionManager.setCurrentThreadContext(new ThreadContext(EXECUTION_OP_ID + "-execution", ThreadType.CONTEXT)); diff --git a/sdk/src/test/java/software/amazon/lambda/durable/context/BaseContextImplTest.java b/sdk/src/test/java/software/amazon/lambda/durable/context/BaseContextImplTest.java index 849f0a1be..56c02f90f 100644 --- a/sdk/src/test/java/software/amazon/lambda/durable/context/BaseContextImplTest.java +++ b/sdk/src/test/java/software/amazon/lambda/durable/context/BaseContextImplTest.java @@ -45,7 +45,8 @@ private ExecutionManager createExecutionManager() { + EXECUTION_NAME + "/" + INVOCATION_ID, "test-token", initialState), - DurableConfig.builder().withDurableExecutionClient(client).build()); + DurableConfig.builder().withDurableExecutionClient(client).build(), + null); } @Test diff --git a/sdk/src/test/java/software/amazon/lambda/durable/context/DurableContextImplTest.java b/sdk/src/test/java/software/amazon/lambda/durable/context/DurableContextImplTest.java index 2d6f6c4e3..2afa7af84 100644 --- a/sdk/src/test/java/software/amazon/lambda/durable/context/DurableContextImplTest.java +++ b/sdk/src/test/java/software/amazon/lambda/durable/context/DurableContextImplTest.java @@ -43,7 +43,8 @@ void setUp() { + EXECUTION_NAME + "/" + INVOCATION_ID, "test-token", initialState), - DurableConfig.builder().withDurableExecutionClient(client).build()); + DurableConfig.builder().withDurableExecutionClient(client).build(), + null); // Simulate the root thread context as the executor would set it executionManager.setCurrentThreadContext(new ThreadContext(null, ThreadType.CONTEXT)); rootContext = DurableContextImpl.createRootContext( 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 3d24aa95b..dea26d90b 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 @@ -33,7 +33,8 @@ private ExecutionManager createManager(List operations) { CheckpointUpdatedExecutionState.builder().operations(operations).build(); return new ExecutionManager( new DurableExecutionInput(EXECUTION_ARN, "test-token", initialState), - DurableConfig.builder().withDurableExecutionClient(client).build()); + DurableConfig.builder().withDurableExecutionClient(client).build(), + null); } private Operation executionOp() { @@ -134,7 +135,8 @@ void emptyInitialState() { .build(); var executionManager = new ExecutionManager( new DurableExecutionInput(EXECUTION_ARN, "test-token", initialState), - DurableConfig.builder().withDurableExecutionClient(client).build()); + DurableConfig.builder().withDurableExecutionClient(client).build(), + null); assertNotNull(executionManager.getExecutionOperation()); assertEquals(EXECUTION_OP_ID, executionManager.getExecutionOperation().id()); @@ -159,7 +161,8 @@ void isOperationUpdatedSinceLastInvocation_returnsTrue_whenOperationIsInList() { var input = new DurableExecutionInput(EXECUTION_ARN, "test-token", initialState, List.of("1")); var manager = new ExecutionManager( input, - DurableConfig.builder().withDurableExecutionClient(client).build()); + DurableConfig.builder().withDurableExecutionClient(client).build(), + null); assertTrue(manager.isOperationUpdatedSinceLastInvocation("1")); } @@ -173,7 +176,8 @@ void isOperationUpdatedSinceLastInvocation_returnsFalse_whenOperationNotInList() var input = new DurableExecutionInput(EXECUTION_ARN, "test-token", initialState, List.of("2")); var manager = new ExecutionManager( input, - DurableConfig.builder().withDurableExecutionClient(client).build()); + DurableConfig.builder().withDurableExecutionClient(client).build(), + null); assertFalse(manager.isOperationUpdatedSinceLastInvocation("1")); } @@ -187,7 +191,8 @@ void isOperationUpdatedSinceLastInvocation_handlesMultipleIds() { var input = new DurableExecutionInput(EXECUTION_ARN, "test-token", initialState, List.of("1", "2", "3")); var manager = new ExecutionManager( input, - DurableConfig.builder().withDurableExecutionClient(client).build()); + DurableConfig.builder().withDurableExecutionClient(client).build(), + null); assertTrue(manager.isOperationUpdatedSinceLastInvocation("1")); assertTrue(manager.isOperationUpdatedSinceLastInvocation("2")); diff --git a/sdk/src/test/java/software/amazon/lambda/durable/operation/CallbackOperationTest.java b/sdk/src/test/java/software/amazon/lambda/durable/operation/CallbackOperationTest.java index f7e5de707..c8cea7934 100644 --- a/sdk/src/test/java/software/amazon/lambda/durable/operation/CallbackOperationTest.java +++ b/sdk/src/test/java/software/amazon/lambda/durable/operation/CallbackOperationTest.java @@ -95,7 +95,8 @@ private ExecutionManager createExecutionManager(List initialOperation CheckpointUpdatedExecutionState.builder().operations(operations).build(); var executionManager = new ExecutionManager( new DurableExecutionInput(EXECUTION_ARN, "test-token", initialState), - DurableConfig.builder().withDurableExecutionClient(client).build()); + DurableConfig.builder().withDurableExecutionClient(client).build(), + null); executionManager.setCurrentThreadContext(new ThreadContext("Root", ThreadType.CONTEXT)); return executionManager; } diff --git a/sdk/src/test/java/software/amazon/lambda/durable/plugin/PluginRunnerTest.java b/sdk/src/test/java/software/amazon/lambda/durable/plugin/PluginRunnerTest.java index faeffb4aa..76d85389d 100644 --- a/sdk/src/test/java/software/amazon/lambda/durable/plugin/PluginRunnerTest.java +++ b/sdk/src/test/java/software/amazon/lambda/durable/plugin/PluginRunnerTest.java @@ -7,6 +7,7 @@ import java.time.Instant; import java.util.ArrayList; import java.util.List; +import java.util.Map; import org.junit.jupiter.api.Test; class PluginRunnerTest { @@ -73,6 +74,7 @@ void fireAndForget_callsAllHookTypes() { runner.onInvocationEnd(invocationEndInfo()); runner.onOperationStart(operationInfo()); runner.onOperationEnd(operationEndInfo()); + runner.onOperationChange(operationChangeInfo()); runner.onUserFunctionStart(attemptInfo()); runner.onUserFunctionEnd(attemptEndInfo()); @@ -82,6 +84,7 @@ void fireAndForget_callsAllHookTypes() { "p:onInvocationEnd", "p:onOperationStart", "p:onOperationEnd", + "p:onOperationChange", "p:onUserFunctionStart", "p:onUserFunctionEnd"), calls); @@ -150,6 +153,11 @@ private static OperationEndInfo operationEndInfo() { "op-1", "test-step", "STEP", null, null, Instant.now(), Instant.now(), "SUCCEEDED", false, null); } + private static OperationChangeInfo operationChangeInfo() { + return new OperationChangeInfo( + "req-123", "arn:aws:lambda:us-east-1:123456789012:function:test", Map.of(), Map.of()); + } + private static UserFunctionStartInfo attemptInfo() { return new UserFunctionStartInfo("op-1", "test-step", "STEP", null, null, Instant.now(), false, 1); } @@ -191,6 +199,11 @@ public void onOperationEnd(OperationEndInfo info) { calls.add(name + ":onOperationEnd"); } + @Override + public void onOperationChange(OperationChangeInfo info) { + calls.add(name + ":onOperationChange"); + } + @Override public void onUserFunctionStart(UserFunctionStartInfo info) { calls.add(name + ":onUserFunctionStart");