Fix getResult hang after termination using test service (#700)

Issue #628

Author: Spikhalskiy

…orkflow/TerminatedWorkflowQueryTest.java …ral/workflow/TerminatedWorkflowTest.javatemporal-sdk/src/test/java/io/temporal/workflow/TerminatedWorkflowQueryTest.java renamed to temporal-sdk/src/test/java/io/temporal/workflow/TerminatedWorkflowTest.java temporal-sdk/src/test/java/io/temporal/workflow/TerminatedWorkflowQueryTest.java renamed to temporal-sdk/src/test/java/io/temporal/workflow/TerminatedWorkflowTest.java

@@ -19,9 +19,10 @@
 
 package io.temporal.workflow;
 
-import io.temporal.client.WorkflowClientOptions;
-import io.temporal.client.WorkflowFailedException;
-import io.temporal.client.WorkflowOptions;
+import static org.junit.Assert.*;
+
+import io.temporal.client.*;
+import io.temporal.failure.TerminatedFailure;
 import io.temporal.testing.internal.SDKTestOptions;
 import io.temporal.testing.internal.SDKTestWorkflowRule;
 import io.temporal.workflow.shared.TestActivities.TestActivitiesImpl;
@@ -30,18 +31,22 @@
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import org.junit.Assert;
 import org.junit.Rule;
 import org.junit.Test;
 
-public class TerminatedWorkflowQueryTest {
-
+/**
+ * Tests verifying the correct behavior of the SDK if the workflow is in unsuccessful final states
+ */
+public class TerminatedWorkflowTest {
   private final TestActivitiesImpl activitiesImpl = new TestActivitiesImpl();
 
   @Rule
   public SDKTestWorkflowRule testWorkflowRule =
       SDKTestWorkflowRule.newBuilder()
-          .setWorkflowTypes(TestTraceWorkflowImpl.class)
+          .setWorkflowTypes(TraceTimingOutWorkflowImpl.class)
           .setActivityImplementations(activitiesImpl)
           .setWorkflowClientOptions(WorkflowClientOptions.newBuilder().build())
           .build();
@@ -53,24 +58,50 @@ public void testShouldReturnQueryResultAfterWorkflowTimeout() {
             .toBuilder()
             .setWorkflowRunTimeout(Duration.ofSeconds(1))
             .build();
-    TestTraceWorkflow client =
+    TestTraceWorkflow workflow =
         testWorkflowRule.getWorkflowClient().newWorkflowStub(TestTraceWorkflow.class, options);
 
     Assert.assertThrows(
         "Workflow should throw because of timeout",
         WorkflowFailedException.class,
-        () -> client.execute(SDKTestWorkflowRule.useExternalService));
+        workflow::execute);
 
-    Assert.assertEquals(1, client.getTrace().size());
-    Assert.assertEquals("started", client.getTrace().get(0));
+    Assert.assertEquals(1, workflow.getTrace().size());
+    Assert.assertEquals("started", workflow.getTrace().get(0));
   }
 
-  public static class TestTraceWorkflowImpl implements TestTraceWorkflow {
+  @Test
+  public void getResultShouldThrowAfterTerminationOfWorkflow() {
+    WorkflowOptions options =
+        WorkflowOptions.newBuilder().setTaskQueue(testWorkflowRule.getTaskQueue()).build();
+
+    WorkflowStub workflow =
+        testWorkflowRule.getWorkflowClient().newUntypedWorkflowStub("execute", options);
+
+    workflow.start();
+
+    workflow.terminate("testing");
+
+    WorkflowFailedException exception = null;
+    try {
+      workflow.getResult(1000, TimeUnit.MILLISECONDS, String.class);
+      fail("getResult should throw WorkflowFailedException because the workflow was terminated");
+    } catch (WorkflowFailedException e) {
+      // This is expected
+      exception = e;
+    } catch (TimeoutException e) {
+      fail(
+          "getResult shouldn't wait all 5 seconds till the end of the workflow because it was already terminated");
+    }
+    assertNotNull(exception);
+    assertTrue(exception.getCause() instanceof TerminatedFailure);
+  }
 
+  public static class TraceTimingOutWorkflowImpl implements TestTraceWorkflow {
     private final List<String> trace = new ArrayList<>();
 
     @Override
-    public String execute(boolean useExternalService) {
+    public String execute() {
       VariousTestActivities localActivities =
           Workflow.newLocalActivityStub(
               VariousTestActivities.class, SDKTestOptions.newLocalActivityOptions());

temporal-sdk/src/test/java/io/temporal/workflow/TimerTest.java

@@ -52,7 +52,7 @@ public void testTimer() {
     }
     TestTraceWorkflow client =
         testWorkflowRule.getWorkflowClient().newWorkflowStub(TestTraceWorkflow.class, options);
-    String result = client.execute(SDKTestWorkflowRule.useExternalService);
+    String result = client.execute();
     Assert.assertEquals("testTimer", result);
     if (testWorkflowRule.isUseExternalService()) {
       testWorkflowRule
@@ -88,9 +88,10 @@ public void testTimer() {
   public static class TestTimerWorkflowImpl implements TestTraceWorkflow {
 
     @Override
-    public String execute(boolean useExternalService) {
+    public String execute() {
       Promise<Void> timer1;
       Promise<Void> timer2;
+      boolean useExternalService = SDKTestWorkflowRule.useExternalService;
       Duration timeout1 = useExternalService ? Duration.ofMillis(700) : Duration.ofSeconds(700);
       Duration timeout2 = useExternalService ? Duration.ofMillis(1300) : Duration.ofSeconds(1300);
       timer1 = Workflow.newTimer(timeout1);

temporal-sdk/src/test/java/io/temporal/workflow/activityTests/AsyncRetryOptionsChangeTest.java

@@ -55,7 +55,7 @@ public void testAsyncRetryOptionsChange() {
         testWorkflowRule.newWorkflowStubTimeoutOptions(TestTraceWorkflow.class);
     String result = null;
     try {
-      result = client.execute(SDKTestWorkflowRule.useExternalService);
+      result = client.execute();
       Assert.fail("unreachable");
     } catch (WorkflowException e) {
       Assert.assertTrue(e.getCause() instanceof ApplicationFailure);
@@ -79,7 +79,7 @@ public static class TestAsyncRetryOptionsChangeWorkflow implements TestTraceWork
     private final List<String> trace = new ArrayList<>();
 
     @Override
-    public String execute(boolean useExternalService) {
+    public String execute() {
       RetryOptions retryOptions;
       if (Workflow.isReplaying()) {
         retryOptions =

temporal-sdk/src/test/java/io/temporal/workflow/activityTests/AsyncRetryTest.java

@@ -48,7 +48,7 @@ public void testAsyncRetry() {
         testWorkflowRule.newWorkflowStubTimeoutOptions(TestTraceWorkflow.class);
     String result = null;
     try {
-      result = client.execute(SDKTestWorkflowRule.useExternalService);
+      result = client.execute();
       Assert.fail("unreachable");
     } catch (WorkflowException e) {
       Assert.assertTrue(e.getCause() instanceof ApplicationFailure);
@@ -76,7 +76,7 @@ public static class TestAsyncRetryWorkflowImpl implements TestTraceWorkflow {
     private final List<String> trace = new ArrayList<>();
 
     @Override
-    public String execute(boolean useExternalService) {
+    public String execute() {
       trace.clear(); // clear because of replay
       trace.add("started");
       Async.retry(

temporal-sdk/src/test/java/io/temporal/workflow/shared/TestWorkflows.java

@@ -114,8 +114,8 @@ public interface ITestNamedChild {
   @WorkflowInterface
   public interface TestTraceWorkflow {
 
-    @WorkflowMethod(name = "testActivity")
-    String execute(boolean useExternalService);
+    @WorkflowMethod(name = "execute")
+    String execute();
 
     @QueryMethod(name = "getTrace")
     List<String> getTrace();

temporal-testing/src/main/java/io/temporal/internal/testservice/StateUtils.java

@@ -0,0 +1,33 @@
+/*
+ *  Copyright (C) 2020 Temporal Technologies, Inc. All Rights Reserved.
+ *
+ *  Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *  Modifications copyright (C) 2017 Uber Technologies, Inc.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License"). You may not
+ *  use this file except in compliance with the License. A copy of the License is
+ *  located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ *  or in the "license" file accompanying this file. This file is distributed on
+ *  an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ *  express or implied. See the License for the specific language governing
+ *  permissions and limitations under the License.
+ */
+
+package io.temporal.internal.testservice;
+
+public class StateUtils {
+  /** @return true if the workflow was completed not by workflow task completion result */
+  public static boolean isWorkflowExecutionForcefullyCompleted(StateMachines.State state) {
+    switch (state) {
+      case TERMINATED:
+      case TIMED_OUT:
+        return true;
+      default:
+        return false;
+    }
+  }
+}

temporal-testing/src/main/java/io/temporal/internal/testservice/TestWorkflowMutableStateImpl.java

@@ -324,11 +324,23 @@ private void update(boolean completeWorkflowTaskUpdate, UpdateProcedure updater,
 
       RequestContext ctx = new RequestContext(clock, this, nextEventId);
       updater.apply(ctx);
-      if (concurrentWorkflowTask && workflow.getState() != State.TIMED_OUT) {
+
+      if (StateUtils.isWorkflowExecutionForcefullyCompleted(workflow.getState())) {
+        // if we completed the workflow "externally", not through the result of workflow task
+        // (timed out or got a termination request) -
+        // we don't buffer the events and don't wait till the finish of the workflow task
+        // in-progress even if there is one,
+        // but instead we apply them to the history immediately.
+        nextEventId = ctx.commitChanges(store);
+      } else if (concurrentWorkflowTask) {
+        // if there is a concurrent workflow task in progress and the workflow wasn't terminated and
+        // considered timed out,
+        // we buffer the events and wait till the finish of the workflow task
         workflowTaskStateMachine.getData().bufferedEvents.add(ctx);
         ctx.fireCallbacks(0);
         store.applyTimersAndLocks(ctx);
       } else {
+        // if there is no concurrent workflow task in progress - apply events to the history
         nextEventId = ctx.commitChanges(store);
       }
     } catch (StatusRuntimeException e) {