Revert "Implement suspend and resume client APIs (#104)"

This reverts commit c303a12.

Author: kaibocai

.github/workflows/build-validation.yml

@@ -95,4 +95,4 @@ jobs:
         uses: actions/upload-artifact@v2
         with:
           name: Integration test report
-          path: client/build/reports/tests/endToEndTest
+          path: client/build/reports/tests/endToEndTest

CHANGELOG.md

@@ -1,7 +1,6 @@
 ## v1.1.0
 
 ### Updates
-* Support Suspend and Resume Client APIs ([#104](https://github.com/microsoft/durabletask-java/issues/104))
 * Fix the potential NPE issue of `DurableTaskClient terminate` method ([#104](https://github.com/microsoft/durabletask-java/issues/104))
 * Add waitForCompletionOrCreateCheckStatusResponse client API ([#115](https://github.com/microsoft/durabletask-java/pull/115))
 * Support long timers by breaking up into smaller timers ([#114](https://github.com/microsoft/durabletask-java/issues/114))

client/src/main/java/com/microsoft/durabletask/DurableTaskClient.java

@@ -281,34 +281,4 @@ public abstract OrchestrationMetadata waitForInstanceCompletion(
      * @return the result of the purge operation, including the number of purged orchestration instances (0 or 1)
      */
     public abstract PurgeResult purgeInstances(PurgeInstanceCriteria purgeInstanceCriteria) throws TimeoutException;
-
-    /**
-     * Suspends a running orchestration instance.
-     * @param instanceId the ID of the orchestration instance to suspend
-     */
-    public void suspendInstance (String instanceId) {
-        this.suspendInstance(instanceId, null);
-    }
-
-    /**
-     * Resumes a running orchestration instance.
-     * @param instanceId the ID of the orchestration instance to resume
-     */
-    public void resumeInstance(String instanceId) {
-        this.resumeInstance(instanceId, null);
-    }
-
-    /**
-     * Suspends a running orchestration instance.
-     * @param instanceId the ID of the orchestration instance to suspend
-     * @param reason the reason for suspending the orchestration instance
-     */
-    public abstract void suspendInstance(String instanceId, @Nullable String reason);
-
-    /**
-     * Resumes a running orchestration instance.
-     * @param instanceId the ID of the orchestration instance to resume
-     * @param reason the reason for resuming the orchestration instance
-     */
-    public abstract void resumeInstance(String instanceId, @Nullable String reason);
 }

client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcClient.java

@@ -204,10 +204,7 @@ public void terminate(String instanceId, @Nullable Object output) {
                 "Terminating instance %s and setting output to: %s",
                 instanceId,
                 serializeOutput != null ? serializeOutput : "(null)"));
-        TerminateRequest.Builder builder = TerminateRequest.newBuilder().setInstanceId(instanceId);
-        if (serializeOutput != null){
-            builder.setOutput(StringValue.of(serializeOutput));
-        }
+        TerminateRequest.Builder builder = TerminateRequest.newBuilder().setInstanceId(instanceId).setOutput(StringValue.of(serializeOutput));
         this.sidecarClient.terminateInstance(builder.build());
     }
 
@@ -283,26 +280,6 @@ public PurgeResult purgeInstances(PurgeInstanceCriteria purgeInstanceCriteria) t
         }
     }
 
-    @Override
-    public void suspendInstance(String instanceId, @Nullable String reason) {
-        SuspendRequest.Builder suspendRequestBuilder = SuspendRequest.newBuilder();
-        suspendRequestBuilder.setInstanceId(instanceId);
-        if (reason != null) {
-            suspendRequestBuilder.setReason(StringValue.of(reason));
-        }
-        this.sidecarClient.suspendInstance(suspendRequestBuilder.build());
-    }
-
-    @Override
-    public void resumeInstance(String instanceId, @Nullable String reason) {
-        ResumeRequest.Builder resumeRequestBuilder = ResumeRequest.newBuilder();
-        resumeRequestBuilder.setInstanceId(instanceId);
-        if (reason != null) {
-            resumeRequestBuilder.setReason(StringValue.of(reason));
-        }
-        this.sidecarClient.resumeInstance(resumeRequestBuilder.build());
-    }
-
     private PurgeResult toPurgeResult(PurgeInstancesResponse response){
         return new PurgeResult(response.getDeletedInstanceCount());
     }

client/src/main/java/com/microsoft/durabletask/OrchestrationRuntimeStatus.java

@@ -46,12 +46,7 @@ public enum OrchestrationRuntimeStatus {
     /**
      * The orchestration was scheduled but hasn't started running.
      */
-    PENDING,
-
-    /**
-     * The orchestration is in a suspended state.
-     */
-    SUSPENDED;
+    PENDING;
 
     static OrchestrationRuntimeStatus fromProtobuf(OrchestrationStatus status) {
         switch (status) {
@@ -69,8 +64,6 @@ static OrchestrationRuntimeStatus fromProtobuf(OrchestrationStatus status) {
                 return TERMINATED;
             case ORCHESTRATION_STATUS_PENDING:
                 return PENDING;
-            case ORCHESTRATION_STATUS_SUSPENDED:
-                return SUSPENDED;
             default:
                 throw new IllegalArgumentException(String.format("Unknown status value: %s", status));
         }
@@ -92,8 +85,6 @@ static OrchestrationStatus toProtobuf(OrchestrationRuntimeStatus status){
                 return ORCHESTRATION_STATUS_TERMINATED;
             case PENDING:
                 return ORCHESTRATION_STATUS_PENDING;
-            case SUSPENDED:
-                return ORCHESTRATION_STATUS_SUSPENDED;
             default:
                 throw new IllegalArgumentException(String.format("Unknown status value: %s", status));
         }

client/src/main/java/com/microsoft/durabletask/TaskOrchestrationExecutor.java

@@ -70,15 +70,13 @@ private class ContextImplTask implements TaskOrchestrationContext {
         private String instanceId;
         private Instant currentInstant;
         private boolean isComplete;
-        private boolean isSuspended;
         private boolean isReplaying = true;
 
         // LinkedHashMap to maintain insertion order when returning the list of pending actions
         private final LinkedHashMap<Integer, OrchestratorAction> pendingActions = new LinkedHashMap<>();
         private final HashMap<Integer, TaskRecord<?>> openTasks = new HashMap<>();
         private final LinkedHashMap<String, Queue<TaskRecord<?>>> outstandingEvents = new LinkedHashMap<>();
         private final LinkedList<HistoryEvent> unprocessedEvents = new LinkedList<>();
-        private final Queue<HistoryEvent> eventsWhileSuspended = new ArrayDeque<>();
         private final DataConverter dataConverter = TaskOrchestrationExecutor.this.dataConverter;
         private final Duration maximumTimerInterval = TaskOrchestrationExecutor.this.maximumTimerInterval;
         private final Logger logger = TaskOrchestrationExecutor.this.logger;
@@ -87,6 +85,7 @@ private class ContextImplTask implements TaskOrchestrationContext {
         private boolean continuedAsNew;
         private Object continuedAsNewInput;
         private boolean preserveUnprocessedEvents;
+
         private Object customStatus;
 
         public ContextImplTask(List<HistoryEvent> pastEvents, List<HistoryEvent> newEvents) {
@@ -530,23 +529,6 @@ private void handleEventRaised(HistoryEvent e) {
             task.complete(result);
         }
 
-        private void handleEventWhileSuspended (HistoryEvent historyEvent){
-            if (historyEvent.getEventTypeCase() != HistoryEvent.EventTypeCase.EXECUTIONSUSPENDED) {
-                eventsWhileSuspended.offer(historyEvent);
-            }
-        }
-
-        private void handleExecutionSuspended(HistoryEvent historyEvent) {
-            this.isSuspended = true;
-        }
-
-        private void handleExecutionResumed(HistoryEvent historyEvent) {
-            this.isSuspended = false;
-            while (!eventsWhileSuspended.isEmpty()) {
-                this.processEvent(eventsWhileSuspended.poll());
-            }
-        }
-
         public Task<Void> createTimer(Duration duration) {
             Helpers.throwIfOrchestratorComplete(this.isComplete);
             Helpers.throwIfArgumentNull(duration, "duration");
@@ -762,86 +744,75 @@ private boolean processNextEvent() {
         }
 
         private void processEvent(HistoryEvent e) {
-            boolean overrideSuspension = e.getEventTypeCase() == HistoryEvent.EventTypeCase.EXECUTIONRESUMED || e.getEventTypeCase() == HistoryEvent.EventTypeCase.EXECUTIONTERMINATED;
-            if (this.isSuspended && !overrideSuspension) {
-                this.handleEventWhileSuspended(e);
-            } else {
-                switch (e.getEventTypeCase()) {
-                    case ORCHESTRATORSTARTED:
-                        Instant instant = DataConverter.getInstantFromTimestamp(e.getTimestamp());
-                        this.setCurrentInstant(instant);
-                        break;
-                    case ORCHESTRATORCOMPLETED:
-                        // No action
-                        break;
-                    case EXECUTIONSTARTED:
-                        ExecutionStartedEvent startedEvent = e.getExecutionStarted();
-                        String name = startedEvent.getName();
-                        this.setName(name);
-                        String instanceId = startedEvent.getOrchestrationInstance().getInstanceId();
-                        this.setInstanceId(instanceId);
-                        String input = startedEvent.getInput().getValue();
-                        this.setInput(input);
-                        TaskOrchestrationFactory factory = TaskOrchestrationExecutor.this.orchestrationFactories.get(name);
-                        if (factory == null) {
-                            // Try getting the default orchestrator
-                            factory = TaskOrchestrationExecutor.this.orchestrationFactories.get("*");
-                        }
-                        // TODO: Throw if the factory is null (orchestration by that name doesn't exist)
-                        TaskOrchestration orchestrator = factory.create();
-                        orchestrator.run(this);
-                        break;
+            switch (e.getEventTypeCase()) {
+                case ORCHESTRATORSTARTED:
+                    Instant instant = DataConverter.getInstantFromTimestamp(e.getTimestamp());
+                    this.setCurrentInstant(instant);
+                    break;
+                case ORCHESTRATORCOMPLETED:
+                    // No action
+                    break;
+                case EXECUTIONSTARTED:
+                    ExecutionStartedEvent startedEvent = e.getExecutionStarted();
+                    String name = startedEvent.getName();
+                    this.setName(name);
+                    String instanceId = startedEvent.getOrchestrationInstance().getInstanceId();
+                    this.setInstanceId(instanceId);
+                    String input = startedEvent.getInput().getValue();
+                    this.setInput(input);
+                    TaskOrchestrationFactory factory = TaskOrchestrationExecutor.this.orchestrationFactories.get(name);
+                    if (factory == null) {
+                        // Try getting the default orchestrator
+                        factory = TaskOrchestrationExecutor.this.orchestrationFactories.get("*");
+                    }
+                    // TODO: Throw if the factory is null (orchestration by that name doesn't exist)
+                    TaskOrchestration orchestrator = factory.create();
+                    orchestrator.run(this);
+                    break;
 //                case EXECUTIONCOMPLETED:
 //                    break;
 //                case EXECUTIONFAILED:
 //                    break;
-                    case EXECUTIONTERMINATED:
-                        this.handleExecutionTerminated(e);
-                        break;
-                    case TASKSCHEDULED:
-                        this.handleTaskScheduled(e);
-                        break;
-                    case TASKCOMPLETED:
-                        this.handleTaskCompleted(e);
-                        break;
-                    case TASKFAILED:
-                        this.handleTaskFailed(e);
-                        break;
-                    case TIMERCREATED:
-                        this.handleTimerCreated(e);
-                        break;
-                    case TIMERFIRED:
-                        this.handleTimerFired(e);
-                        break;
-                    case SUBORCHESTRATIONINSTANCECREATED:
-                        this.handleSubOrchestrationCreated(e);
-                        break;
-                    case SUBORCHESTRATIONINSTANCECOMPLETED:
-                        this.handleSubOrchestrationCompleted(e);
-                        break;
-                    case SUBORCHESTRATIONINSTANCEFAILED:
-                        this.handleSubOrchestrationFailed(e);
-                        break;
+                case EXECUTIONTERMINATED:
+                    this.handleExecutionTerminated(e);
+                    break;
+                case TASKSCHEDULED:
+                    this.handleTaskScheduled(e);
+                    break;
+                case TASKCOMPLETED:
+                    this.handleTaskCompleted(e);
+                    break;
+                case TASKFAILED:
+                    this.handleTaskFailed(e);
+                    break;
+                case TIMERCREATED:
+                    this.handleTimerCreated(e);
+                    break;
+                case TIMERFIRED:
+                    this.handleTimerFired(e);
+                    break;
+                case SUBORCHESTRATIONINSTANCECREATED:
+                    this.handleSubOrchestrationCreated(e);
+                    break;
+                case SUBORCHESTRATIONINSTANCECOMPLETED:
+                    this.handleSubOrchestrationCompleted(e);
+                    break;
+                case SUBORCHESTRATIONINSTANCEFAILED:
+                    this.handleSubOrchestrationFailed(e);
+                    break;
 //                case EVENTSENT:
 //                    break;
-                    case EVENTRAISED:
-                        this.handleEventRaised(e);
-                        break;
+                case EVENTRAISED:
+                    this.handleEventRaised(e);
+                    break;
 //                case GENERICEVENT:
 //                    break;
 //                case HISTORYSTATE:
 //                    break;
 //                case EVENTTYPE_NOT_SET:
 //                    break;
-                    case EXECUTIONSUSPENDED:
-                        this.handleExecutionSuspended(e);
-                        break;
-                    case EXECUTIONRESUMED:
-                        this.handleExecutionResumed(e);
-                        break;
-                    default:
-                        throw new IllegalStateException("Don't know how to handle history type " + e.getEventTypeCase());
-                }
+                default:
+                    throw new IllegalStateException("Don't know how to handle history type " + e.getEventTypeCase());
             }
         }
 

client/src/test/java/com/microsoft/durabletask/IntegrationTests.java

@@ -408,72 +408,6 @@ void termination() throws TimeoutException {
         }
     }
 
-    @Test
-    void suspendResumeOrchestration() throws TimeoutException, InterruptedException {
-        final String orchestratorName = "suspend";
-        final String eventName = "MyEvent";
-        final String eventPayload = "testPayload";
-        final Duration suspendTimeout = Duration.ofSeconds(5);
-
-        DurableTaskGrpcWorker worker = this.createWorkerBuilder()
-                .addOrchestrator(orchestratorName, ctx -> {
-                    String payload = ctx.waitForExternalEvent(eventName, String.class).await();
-                    ctx.complete(payload);
-                })
-                .buildAndStart();
-
-        DurableTaskClient client = new DurableTaskGrpcClientBuilder().build();
-        try (worker; client) {
-            String instanceId = client.scheduleNewOrchestrationInstance(orchestratorName);
-            client.suspendInstance(instanceId);
-            OrchestrationMetadata instance = client.waitForInstanceStart(instanceId, defaultTimeout);
-            assertNotNull(instance);
-            assertEquals(OrchestrationRuntimeStatus.SUSPENDED, instance.getRuntimeStatus());
-
-            client.raiseEvent(instanceId, eventName, eventPayload);
-
-            assertThrows(
-                    TimeoutException.class,
-                    () -> client.waitForInstanceCompletion(instanceId, suspendTimeout, false),
-                    "Expected to throw TimeoutException, but it didn't"
-                    );
-
-            String resumeReason = "Resume for testing.";
-            client.resumeInstance(instanceId, resumeReason);
-            instance = client.waitForInstanceCompletion(instanceId, defaultTimeout, true);
-            assertNotNull(instance);
-            assertEquals(instanceId, instance.getInstanceId());
-            assertEquals(eventPayload, instance.readOutputAs(String.class));
-            assertEquals(OrchestrationRuntimeStatus.COMPLETED, instance.getRuntimeStatus());
-        }
-    }
-
-    @Test
-    void terminateSuspendOrchestration() throws TimeoutException, InterruptedException {
-        final String orchestratorName = "suspendResume";
-        final String eventName = "MyEvent";
-        final String eventPayload = "testPayload";
-
-        DurableTaskGrpcWorker worker = this.createWorkerBuilder()
-                .addOrchestrator(orchestratorName, ctx -> {
-                    String payload = ctx.waitForExternalEvent(eventName, String.class).await();
-                    ctx.complete(payload);
-                })
-                .buildAndStart();
-
-        DurableTaskClient client = new DurableTaskGrpcClientBuilder().build();
-        try (worker; client) {
-            String instanceId = client.scheduleNewOrchestrationInstance(orchestratorName);
-            String suspendReason = "Suspend for testing.";
-            client.suspendInstance(instanceId, suspendReason);
-            client.terminate(instanceId, null);
-            OrchestrationMetadata instance = client.waitForInstanceCompletion(instanceId, defaultTimeout, false);
-            assertNotNull(instance);
-            assertEquals(instanceId, instance.getInstanceId());
-            assertEquals(OrchestrationRuntimeStatus.TERMINATED, instance.getRuntimeStatus());
-        }
-    }
-
     @Test
     void activityFanOut() throws IOException, TimeoutException {
         final String orchestratorName = "ActivityFanOut";