Make sure signals are applied in the same order as they appear in history (#318)

* Make sure signals in the same decision are applied in the same order as they appear in history
* Support suspend/resume polling methods in worker factory

Author: meiliang86

src/main/java/com/uber/cadence/internal/sync/DeterministicRunnerImpl.java

@@ -202,23 +202,34 @@ public void runUntilAllBlocked() throws Throwable {
       outerLoop:
       do {
         threadsToAdd.clear();
-        for (NamedRunnable nr : toExecuteInWorkflowThread) {
-          WorkflowThread thread =
-              new WorkflowThreadImpl(
-                  false,
-                  threadPool,
-                  this,
-                  nr.name,
-                  false,
-                  runnerCancellationScope,
-                  nr.runnable,
-                  cache);
+
+        if (!toExecuteInWorkflowThread.isEmpty()) {
+          List<WorkflowThread> callbackThreads = new ArrayList<>(toExecuteInWorkflowThread.size());
+          for (NamedRunnable nr : toExecuteInWorkflowThread) {
+            WorkflowThread thread =
+                new WorkflowThreadImpl(
+                    false,
+                    threadPool,
+                    this,
+                    nr.name,
+                    false,
+                    runnerCancellationScope,
+                    nr.runnable,
+                    cache);
+            callbackThreads.add(thread);
+          }
+
           // It is important to prepend threads as there are callbacks
           // like signals that have to run before any other threads.
           // Otherwise signal might be never processed if it was received
           // after workflow decided to close.
-          threads.addFirst(thread);
+          // Adding the callbacks in the same order as they appear in history.
+
+          for (int i = callbackThreads.size() - 1; i >= 0; i--) {
+            threads.addFirst(callbackThreads.get(i));
+          }
         }
+
         toExecuteInWorkflowThread.clear();
         progress = false;
         Iterator<WorkflowThread> ci = threads.iterator();

src/main/java/com/uber/cadence/internal/sync/SyncActivityWorker.java

@@ -19,15 +19,15 @@
 
 import com.uber.cadence.internal.common.InternalUtils;
 import com.uber.cadence.internal.worker.ActivityWorker;
-import com.uber.cadence.internal.worker.Lifecycle;
 import com.uber.cadence.internal.worker.SingleWorkerOptions;
+import com.uber.cadence.internal.worker.SuspendableWorker;
 import com.uber.cadence.serviceclient.IWorkflowService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 
 /** Activity worker that supports POJO activity implementations. */
-public class SyncActivityWorker implements Lifecycle {
+public class SyncActivityWorker implements SuspendableWorker {
 
   private final ActivityWorker worker;
   private final POJOActivityTaskHandler taskHandler;
@@ -83,11 +83,18 @@ public void awaitTermination(long timeout, TimeUnit unit) {
     InternalUtils.awaitTermination(heartbeatExecutor, timeoutMillis);
   }
 
+  @Override
   public void suspendPolling() {
     worker.suspendPolling();
   }
 
+  @Override
   public void resumePolling() {
     worker.resumePolling();
   }
+
+  @Override
+  public boolean isSuspended() {
+    return worker.isSuspended();
+  }
 }

src/main/java/com/uber/cadence/worker/Worker.java

@@ -35,6 +35,7 @@
 import com.uber.cadence.internal.worker.Poller;
 import com.uber.cadence.internal.worker.PollerOptions;
 import com.uber.cadence.internal.worker.SingleWorkerOptions;
+import com.uber.cadence.internal.worker.Suspendable;
 import com.uber.cadence.internal.worker.WorkflowPollTaskFactory;
 import com.uber.cadence.serviceclient.IWorkflowService;
 import com.uber.cadence.serviceclient.WorkflowServiceTChannel;
@@ -63,7 +64,7 @@
  * Hosts activity and workflow implementations. Uses long poll to receive activity and decision
  * tasks and processes them in a correspondent thread pool.
  */
-public final class Worker {
+public final class Worker implements Suspendable {
 
   private final WorkerOptions options;
   private final String taskList;
@@ -364,6 +365,43 @@ public String getTaskList() {
     return taskList;
   }
 
+  @Override
+  public void suspendPolling() {
+    if (workflowWorker != null) {
+      workflowWorker.suspendPolling();
+    }
+
+    if (activityWorker != null) {
+      activityWorker.suspendPolling();
+    }
+  }
+
+  @Override
+  public void resumePolling() {
+    if (workflowWorker != null) {
+      workflowWorker.resumePolling();
+    }
+
+    if (activityWorker != null) {
+      activityWorker.resumePolling();
+    }
+  }
+
+  @Override
+  public boolean isSuspended() {
+    boolean workflowWorkerSuspended = true;
+    if (workflowWorker != null) {
+      workflowWorkerSuspended = workflowWorker.isSuspended();
+    }
+
+    boolean activityWorkerSuspended = activityWorker.isSuspended();
+    if (activityWorker != null) {
+      activityWorker.resumePolling();
+    }
+
+    return workflowWorkerSuspended && activityWorkerSuspended;
+  }
+
   /** Maintains worker creation and lifecycle. */
   public static final class Factory {
     private final List<Worker> workers = new ArrayList<>();
@@ -712,9 +750,40 @@ private String getStickyTaskListName() {
           : String.format("%s:%s", getHostName(), id);
     }
 
+    public synchronized void suspendPolling() {
+      if (state != State.Started) {
+        return;
+      }
+
+      log.info("suspendPolling");
+      state = State.Suspended;
+      if (stickyPoller != null) {
+        stickyPoller.suspendPolling();
+      }
+      for (Worker worker : workers) {
+        worker.suspendPolling();
+      }
+    }
+
+    public synchronized void resumePolling() {
+      if (state != State.Suspended) {
+        return;
+      }
+
+      log.info("resumePolling");
+      state = State.Started;
+      if (stickyPoller != null) {
+        stickyPoller.resumePolling();
+      }
+      for (Worker worker : workers) {
+        worker.resumePolling();
+      }
+    }
+
     enum State {
       Initial,
       Started,
+      Suspended,
       Shutdown
     }
   }

src/test/java/com/uber/cadence/workflow/WorkflowTest.java

@@ -4569,6 +4569,64 @@ public void testParallelLocalActivityExecutionWorkflow() {
         result);
   }
 
+  public interface SignalOrderingWorkflow {
+    @WorkflowMethod
+    List<String> run();
+
+    @SignalMethod(name = "testSignal")
+    void signal(String s);
+  }
+
+  public static class SignalOrderingWorkflowImpl implements SignalOrderingWorkflow {
+    private List<String> signals = new ArrayList<>();
+
+    @Override
+    public List<String> run() {
+      Workflow.await(() -> signals.size() == 3);
+      return signals;
+    }
+
+    @Override
+    public void signal(String s) {
+      signals.add(s);
+    }
+  }
+
+  @Test
+  public void testSignalOrderingWorkflow() {
+    startWorkerFor(SignalOrderingWorkflowImpl.class);
+    WorkflowOptions options =
+        new WorkflowOptions.Builder()
+            .setExecutionStartToCloseTimeout(Duration.ofMinutes(1))
+            .setTaskStartToCloseTimeout(Duration.ofSeconds(10))
+            .setTaskList(taskList)
+            .build();
+    SignalOrderingWorkflow workflowStub =
+        workflowClient.newWorkflowStub(SignalOrderingWorkflow.class, options);
+    WorkflowClient.start(workflowStub::run);
+
+    // Suspend polling so that all the signals will be received in the same decision task.
+    if (useExternalService) {
+      workerFactory.suspendPolling();
+    } else {
+      testEnvironment.getWorkerFactory().suspendPolling();
+    }
+
+    workflowStub.signal("test1");
+    workflowStub.signal("test2");
+    workflowStub.signal("test3");
+
+    if (useExternalService) {
+      workerFactory.resumePolling();
+    } else {
+      testEnvironment.getWorkerFactory().resumePolling();
+    }
+
+    List<String> result = workflowStub.run();
+    List<String> expected = Arrays.asList("test1", "test2", "test3");
+    assertEquals(expected, result);
+  }
+
   private static class FilteredTrace {
 
     private final List<String> impl = Collections.synchronizedList(new ArrayList<>());