Don't schedule local activities on a completed workflow (#1908)

Don't schedule LA on completed workflow

Author: Quinn-With-Two-Ns

.buildkite/pipeline.yml

@@ -4,7 +4,7 @@ steps:
       queue: "default"
       docker: "*"
     command: "./gradlew --no-daemon test -x checkLicenseMain -x checkLicenses -x spotlessCheck -x spotlessApply -x spotlessJava -P edgeDepsTest"
-    timeout_in_minutes: 15
+    timeout_in_minutes: 20
     plugins:
       - docker-compose#v3.8.0:
           run: unit-test-test-service-edge
@@ -15,7 +15,7 @@ steps:
       queue: "default"
       docker: "*"
     command: "./gradlew --no-daemon test -x checkLicenseMain -x checkLicenses -x spotlessCheck -x spotlessApply -x spotlessJava"
-    timeout_in_minutes: 15
+    timeout_in_minutes: 20
     plugins:
       - docker-compose#v3.8.0:
           run: unit-test-docker-jdk8
@@ -26,7 +26,7 @@ steps:
       queue: "default"
       docker: "*"
     command: "./gradlew --no-daemon checkLicenseMain checkLicenses spotlessCheck"
-    timeout_in_minutes: 15
+    timeout_in_minutes: 20
     plugins:
       - docker-compose#v3.8.0:
           run: jdk11

temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflowRunTaskHandler.java

@@ -311,6 +311,12 @@ private void processLocalActivityRequests(Deadline wftHeartbeatDeadline)
       throws InterruptedException, Throwable {
 
     while (true) {
+      // Scheduling or handling any local activities after the workflow method has returned
+      // can result in commands being generated after the CompleteWorkflowExecution command
+      // which the server does not allow.
+      if (context.isWorkflowMethodCompleted()) {
+        break;
+      }
       List<ExecuteLocalActivityParameters> laRequests =
           workflowStateMachines.takeLocalActivityRequests();
       localActivityTaskCount += laRequests.size();
@@ -361,7 +367,8 @@ private void processLocalActivityRequests(Deadline wftHeartbeatDeadline)
     // it's safe to call and discard the result of takeLocalActivityRequests() here, because if it's
     // not empty - we are in trouble anyway
     Preconditions.checkState(
-        workflowStateMachines.takeLocalActivityRequests().isEmpty(),
+        workflowStateMachines.takeLocalActivityRequests().isEmpty()
+            || context.isWorkflowMethodCompleted(),
         "[BUG] Local activities requests from the last event loop were not drained "
             + "and accounted in the outstanding local activities counter");
   }

temporal-sdk/src/main/java/io/temporal/internal/sync/SyncWorkflow.java

@@ -161,9 +161,6 @@ public void handleUpdate(
           // Skip validator on replay
           if (!callbacks.isReplaying()) {
             try {
-              // TODO(https://github.com/temporalio/sdk-java/issues/1748) handleValidateUpdate
-              // should not just be run
-              // in a workflow thread
               workflowContext.setReadOnly(true);
               workflowProc.handleValidateUpdate(updateName, input, eventId, header);
             } catch (Exception e) {

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

@@ -0,0 +1,59 @@
+/*
+ * Copyright (C) 2022 Temporal Technologies, Inc. All Rights Reserved.
+ *
+ * Copyright (C) 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 material except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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.workflow;
+
+import static org.junit.Assert.assertEquals;
+
+import io.temporal.testing.internal.SDKTestWorkflowRule;
+import io.temporal.workflow.shared.TestActivities;
+import io.temporal.workflow.shared.TestWorkflows;
+import java.util.Objects;
+import org.junit.Rule;
+import org.junit.Test;
+
+public class CommandInTheLastWorkflowTaskTest {
+  @Rule
+  public SDKTestWorkflowRule testWorkflowRule =
+      SDKTestWorkflowRule.newBuilder()
+          .setWorkflowTypes(TestWorkflowImpl.class)
+          .setActivityImplementations(new TestActivities.TestActivitiesImpl())
+          .build();
+
+  @Test
+  public void testCommandInTheLastWorkflowTask() {
+    TestWorkflows.TestWorkflowReturnString client =
+        testWorkflowRule.newWorkflowStub(TestWorkflows.TestWorkflowReturnString.class);
+    assertEquals("done", client.execute());
+  }
+
+  public static class TestWorkflowImpl implements TestWorkflows.TestWorkflowReturnString {
+
+    @Override
+    public String execute() {
+      Async.procedure(
+          () -> {
+            Workflow.mutableSideEffect(
+                "id1", Integer.class, (o, n) -> Objects.equals(n, o), () -> 0);
+          });
+      return "done";
+    }
+  }
+}

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

@@ -0,0 +1,65 @@
+/*
+ * Copyright (C) 2022 Temporal Technologies, Inc. All Rights Reserved.
+ *
+ * Copyright (C) 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 material except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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.workflow.activityTests;
+
+import static org.junit.Assert.assertEquals;
+
+import io.temporal.activity.ActivityOptions;
+import io.temporal.testing.internal.SDKTestWorkflowRule;
+import io.temporal.workflow.Async;
+import io.temporal.workflow.Workflow;
+import io.temporal.workflow.shared.TestActivities;
+import io.temporal.workflow.shared.TestWorkflows;
+import java.time.Duration;
+import org.junit.Rule;
+import org.junit.Test;
+
+public class ActivityInTheLastWorkflowTaskTest {
+  @Rule
+  public SDKTestWorkflowRule testWorkflowRule =
+      SDKTestWorkflowRule.newBuilder()
+          .setWorkflowTypes(TestWorkflowImpl.class)
+          .setActivityImplementations(new TestActivities.TestActivitiesImpl())
+          .build();
+
+  @Test
+  public void testActivityInTheLastWorkflowTask() {
+    TestWorkflows.TestWorkflowReturnString client =
+        testWorkflowRule.newWorkflowStub(TestWorkflows.TestWorkflowReturnString.class);
+    assertEquals("done", client.execute());
+  }
+
+  public static class TestWorkflowImpl implements TestWorkflows.TestWorkflowReturnString {
+
+    private final TestActivities.VariousTestActivities activities =
+        Workflow.newActivityStub(
+            TestActivities.VariousTestActivities.class,
+            ActivityOptions.newBuilder()
+                .setScheduleToCloseTimeout(Duration.ofSeconds(200))
+                .build());
+
+    @Override
+    public String execute() {
+      Async.procedure(activities::sleepActivity, (long) 1000, 0);
+      return "done";
+    }
+  }
+}

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

@@ -0,0 +1,78 @@
+/*
+ * Copyright (C) 2022 Temporal Technologies, Inc. All Rights Reserved.
+ *
+ * Copyright (C) 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 material except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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.workflow.activityTests;
+
+import static org.junit.Assert.assertEquals;
+
+import io.temporal.activity.LocalActivityOptions;
+import io.temporal.testing.internal.SDKTestWorkflowRule;
+import io.temporal.workflow.*;
+import io.temporal.workflow.shared.TestActivities;
+import java.time.Duration;
+import junitparams.JUnitParamsRunner;
+import junitparams.Parameters;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+@RunWith(JUnitParamsRunner.class)
+public class LocalActivityInTheLastWorkflowTaskTest {
+  @Rule
+  public SDKTestWorkflowRule testWorkflowRule =
+      SDKTestWorkflowRule.newBuilder()
+          .setWorkflowTypes(TestWorkflowImpl.class)
+          .setActivityImplementations(new TestActivities.TestActivitiesImpl())
+          .build();
+
+  @Test
+  @Parameters({"true", "false"})
+  public void testLocalActivityInTheLastWorkflowTask(boolean blockOnLA) {
+    TestWorkflow client = testWorkflowRule.newWorkflowStub(TestWorkflow.class);
+    assertEquals("done", client.execute(blockOnLA));
+  }
+
+  @WorkflowInterface
+  public interface TestWorkflow {
+    @WorkflowMethod
+    String execute(boolean blockOnLA);
+  }
+
+  public static class TestWorkflowImpl implements TestWorkflow {
+
+    private final TestActivities.VariousTestActivities activities =
+        Workflow.newLocalActivityStub(
+            TestActivities.VariousTestActivities.class,
+            LocalActivityOptions.newBuilder()
+                .setScheduleToCloseTimeout(Duration.ofSeconds(200))
+                .build());
+
+    @Override
+    public String execute(boolean blockOnLA) {
+      if (blockOnLA) {
+        Promise promise = Async.procedure(activities::sleepActivity, (long) 100, 0);
+        Async.procedure(activities::sleepActivity, (long) 1000, 0);
+        promise.get();
+      }
+      Async.procedure(activities::sleepActivity, (long) 1000, 0);
+      return "done";
+    }
+  }
+}

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

@@ -177,6 +177,16 @@ public interface QueryableWorkflow {
     void mySignal(String value);
   }
 
+  @WorkflowInterface
+  public interface SimpleWorkflowWithUpdate {
+
+    @WorkflowMethod
+    String execute();
+
+    @UpdateMethod
+    String update(String value);
+  }
+
   @WorkflowInterface
   public interface WorkflowWithUpdate {
 

temporal-sdk/src/test/java/io/temporal/workflow/signalTests/SignalWithLocalActivityInTheLastWorkflowTaskTest.java

@@ -0,0 +1,90 @@
+/*
+ * Copyright (C) 2022 Temporal Technologies, Inc. All Rights Reserved.
+ *
+ * Copyright (C) 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 material except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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.workflow.signalTests;
+
+import static org.junit.Assert.assertEquals;
+
+import io.temporal.activity.LocalActivityOptions;
+import io.temporal.client.WorkflowStub;
+import io.temporal.testing.internal.SDKTestWorkflowRule;
+import io.temporal.workflow.*;
+import io.temporal.workflow.shared.TestActivities;
+import java.time.Duration;
+import junitparams.JUnitParamsRunner;
+import junitparams.Parameters;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+@RunWith(JUnitParamsRunner.class)
+public class SignalWithLocalActivityInTheLastWorkflowTaskTest {
+  @Rule
+  public SDKTestWorkflowRule testWorkflowRule =
+      SDKTestWorkflowRule.newBuilder()
+          .setWorkflowTypes(TestSignalWorkflowImpl.class)
+          .setActivityImplementations(new TestActivities.TestActivitiesImpl())
+          .build();
+
+  @Test
+  @Parameters({"true", "false"})
+  public void testSignalWithLocalActivityInTheLastWorkflowTask(Boolean waitOnLA) {
+    TestSignaledWorkflow client = testWorkflowRule.newWorkflowStub(TestSignaledWorkflow.class);
+    WorkflowStub.fromTyped(client)
+        .signalWithStart("testSignal", new String[] {"signalValue"}, new Boolean[] {waitOnLA});
+    assertEquals("done", client.execute());
+  }
+
+  @WorkflowInterface
+  public interface TestSignaledWorkflow {
+
+    @WorkflowMethod
+    String execute();
+
+    @SignalMethod
+    void signal(boolean waitOnLA);
+  }
+
+  public static class TestSignalWorkflowImpl implements TestSignaledWorkflow {
+
+    private final TestActivities.VariousTestActivities activities =
+        Workflow.newLocalActivityStub(
+            TestActivities.VariousTestActivities.class,
+            LocalActivityOptions.newBuilder()
+                .setScheduleToCloseTimeout(Duration.ofSeconds(200))
+                .build());
+
+    @Override
+    public String execute() {
+      return "done";
+    }
+
+    @Override
+    public void signal(boolean waitOnLA) {
+      if (waitOnLA) {
+        Promise promise = Async.procedure(activities::sleepActivity, (long) 100, 0);
+        Async.procedure(activities::sleepActivity, (long) 1000, 0);
+        promise.get();
+      }
+
+      activities.sleepActivity(1000, 0);
+    }
+  }
+}