dedup isTerminalStatus

Author: zhongkechen

sdk/src/main/java/com/amazonaws/lambda/durable/execution/ExecutionManager.java

@@ -208,7 +208,7 @@ public void shutdown() {
         checkpointBatcher.shutdown();
     }
 
-    private boolean isTerminalStatus(OperationStatus status) {
+    public static boolean isTerminalStatus(OperationStatus status) {
         return status == OperationStatus.SUCCEEDED
                 || status == OperationStatus.FAILED
                 || status == OperationStatus.CANCELLED

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

@@ -19,7 +19,6 @@
 import org.slf4j.LoggerFactory;
 import software.amazon.awssdk.services.lambda.model.ErrorObject;
 import software.amazon.awssdk.services.lambda.model.Operation;
-import software.amazon.awssdk.services.lambda.model.OperationStatus;
 import software.amazon.awssdk.services.lambda.model.OperationType;
 import software.amazon.awssdk.services.lambda.model.OperationUpdate;
 
@@ -50,7 +49,7 @@ public abstract class BaseDurableOperation<T> implements DurableFuture<T> {
     private final ExecutionManager executionManager;
     private final TypeToken<T> resultTypeToken;
     private final SerDes resultSerDes;
-    private final CompletableFuture<Void> completionFuture;
+    protected final CompletableFuture<Void> completionFuture;
 
     protected BaseDurableOperation(
             String operationId,
@@ -146,7 +145,9 @@ protected Operation waitForOperationCompletion() {
 
     /** Receives operation updates from ExecutionManager and updates the internal state of the operation */
     public void onCheckpointComplete(Operation operation) {
-        if (isTerminalStatus(operation.status())) {
+        if (ExecutionManager.isTerminalStatus(operation.status())) {
+            // This method handles only terminal status updates. Override this method if a DurableOperation needs to
+            // handle other updates.
             logger.trace("In onCheckpointComplete, completing operation {} ({})", operationId, completionFuture);
             // It's important that we synchronize access to the future, otherwise the processing could happen
             // on someone else's thread and cause a race condition.
@@ -292,12 +293,4 @@ protected void validateReplay(Operation checkpointed) {
                     operationId, checkpointed.name(), getName())));
         }
     }
-
-    private boolean isTerminalStatus(OperationStatus status) {
-        return status == OperationStatus.SUCCEEDED
-                || status == OperationStatus.FAILED
-                || status == OperationStatus.CANCELLED
-                || status == OperationStatus.TIMED_OUT
-                || status == OperationStatus.STOPPED;
-    }
 }