move parentId to base class

Author: zhongkechen

sdk/src/main/java/com/amazonaws/lambda/durable/DurableExecutor.java

@@ -78,8 +78,8 @@ public static <I, O> DurableExecutionOutput execute(
                 },
                 config.getExecutorService()); // Get executor from config for running user code
 
-        // Get suspend future from ExecutionManager. If this future completes, it
-        // indicates that no threads are active and we can safely suspend.
+        // Execute the handlerFuture in ExecutionManager. If it completes successfully, the output of user function
+        // will be returned. Otherwise, it will complete exceptionally with a SuspendExecutionException or a failure.
         return executionManager
                 .execute(handlerFuture)
                 .handle((result, ex) -> {

sdk/src/main/java/com/amazonaws/lambda/durable/operation/BaseDurableOperation.java

@@ -218,8 +218,12 @@ protected void sendOperationUpdate(OperationUpdate.Builder builder) {
     }
 
     protected CompletableFuture<Void> sendOperationUpdateAsync(OperationUpdate.Builder builder) {
-        return executionManager.sendOperationUpdate(
-                builder.id(operationId).name(name).type(operationType).build());
+        // todo: add parentId when we support operations in child context
+        return executionManager.sendOperationUpdate(builder.id(operationId)
+                .name(name)
+                .type(operationType)
+                .parentId(null)
+                .build());
     }
 
     // serialization/deserialization utilities

sdk/src/main/java/com/amazonaws/lambda/durable/operation/CallbackOperation.java

@@ -67,10 +67,8 @@ public void execute() {
             }
         } else {
             // First execution: checkpoint and get callback ID
-            var update = OperationUpdate.builder()
-                    .parentId(null)
-                    .action(OperationAction.START)
-                    .callbackOptions(buildCallbackOptions());
+            var update =
+                    OperationUpdate.builder().action(OperationAction.START).callbackOptions(buildCallbackOptions());
 
             sendOperationUpdate(update);
 

sdk/src/main/java/com/amazonaws/lambda/durable/operation/InvokeOperation.java

@@ -74,7 +74,6 @@ private void waitTimeout() {
 
     private void startInvocation() {
         var update = OperationUpdate.builder()
-                .parentId(null)
                 .action(OperationAction.START)
                 .chainedInvokeOptions(ChainedInvokeOptions.builder()
                         .functionName(functionName)

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

@@ -124,7 +124,7 @@ private void executeStepLogic(int attempt) {
                 // Check if we need to send START
                 var existing = getOperation();
                 if (existing == null || existing.status() != OperationStatus.STARTED) {
-                    var startUpdate = OperationUpdate.builder().parentId(null).action(OperationAction.START);
+                    var startUpdate = OperationUpdate.builder().action(OperationAction.START);
 
                     if (config.semantics() == StepSemantics.AT_MOST_ONCE_PER_RETRY) {
                         // AT_MOST_ONCE: await START checkpoint before executing user code
@@ -140,7 +140,6 @@ private void executeStepLogic(int attempt) {
 
                 // Send SUCCEED
                 var successUpdate = OperationUpdate.builder()
-                        .parentId(null)
                         .action(OperationAction.SUCCEED)
                         .payload(serializeResult(result));
                 sendOperationUpdate(successUpdate);
@@ -180,7 +179,6 @@ private void handleStepFailure(Throwable exception, int attempt) {
         if (isRetryable && retryDecision.shouldRetry()) {
             // Send RETRY
             var retryUpdate = OperationUpdate.builder()
-                    .parentId(null)
                     .action(OperationAction.RETRY)
                     .error(errorObject)
                     .stepOptions(StepOptions.builder()
@@ -199,10 +197,8 @@ private void handleStepFailure(Throwable exception, int attempt) {
             pollUntilReady(pendingFuture, nextAttemptTime, Duration.ofMillis(200));
         } else {
             // Send FAIL - retries exhausted
-            var failUpdate = OperationUpdate.builder()
-                    .parentId(null)
-                    .action(OperationAction.FAIL)
-                    .error(errorObject);
+            var failUpdate =
+                    OperationUpdate.builder().action(OperationAction.FAIL).error(errorObject);
             sendOperationUpdate(failUpdate);
         }
     }

sdk/src/main/java/com/amazonaws/lambda/durable/operation/WaitOperation.java

@@ -50,7 +50,6 @@ public void execute() {
         } else {
             // First execution - checkpoint with full duration
             var update = OperationUpdate.builder()
-                    .parentId(null)
                     .action(OperationAction.START)
                     .waitOptions(WaitOptions.builder()
                             .waitSeconds((int) duration.toSeconds())

sdk/src/test/java/com/amazonaws/lambda/durable/operation/BaseDurableOperationTest.java

@@ -120,7 +120,7 @@ public String get() {
     @Test
     void waitForOperationCompletionWhenRunningAndReadyToComplete() {
         Phaser phaser = new Phaser(0);
-        OperationContext context = new OperationContext("step", ThreadType.STEP);
+        OperationContext context = new OperationContext("step", ThreadType.CONTEXT);
         ExecutionManager executionManager = mock(ExecutionManager.class);
         when(executionManager.getCurrentContext()).thenReturn(context);
         when(executionManager.startPhaser(OPERATION_ID)).thenReturn(phaser);
@@ -151,7 +151,7 @@ public String get() {
     void waitForOperationCompletionWhenAlreadyCompleted() {
         Phaser phaser = new Phaser(1);
         phaser.arrive(); // completed
-        OperationContext context = new OperationContext("step", ThreadType.STEP);
+        OperationContext context = new OperationContext("step", ThreadType.CONTEXT);
         ExecutionManager executionManager = mock(ExecutionManager.class);
         when(executionManager.getCurrentContext()).thenReturn(context);
         when(executionManager.startPhaser(OPERATION_ID)).thenReturn(phaser);