1. Refactor decider (#201)

decider returns list of decisions on decide

Author: halakaraki

src/main/java/com/uber/cadence/internal/replay/Decider.java

@@ -17,13 +17,14 @@
 
 package com.uber.cadence.internal.replay;
 
+import com.uber.cadence.Decision;
 import com.uber.cadence.PollForDecisionTaskResponse;
 import com.uber.cadence.WorkflowQuery;
+import java.util.List;
 
 public interface Decider {
 
-  // TODO: refactor in future CR. Merge methods and decide should return a list of decisions.
-  void decide(PollForDecisionTaskResponse decisionTask) throws Throwable;
+  List<Decision> decide(PollForDecisionTaskResponse decisionTask) throws Throwable;
 
   byte[] query(PollForDecisionTaskResponse decisionTask, WorkflowQuery query) throws Throwable;
 

src/main/java/com/uber/cadence/internal/replay/DeciderCache.java

@@ -110,8 +110,8 @@ public long size() {
 
   private boolean isFullHistory(PollForDecisionTaskResponse decisionTask) {
     return decisionTask.getHistory() != null
-        && decisionTask.getHistory().getEventsSize() > 0
-        && decisionTask.history.events.get(0).getEventId() == 1;
+        && decisionTask.getHistory().getEvents().size() > 0
+        && decisionTask.getHistory().getEvents().get(0).getEventId() == 1;
   }
 
   public void invalidateAll() {

src/main/java/com/uber/cadence/internal/replay/DecisionsHelper.java

@@ -30,7 +30,6 @@
 import com.uber.cadence.CompleteWorkflowExecutionDecisionAttributes;
 import com.uber.cadence.ContinueAsNewWorkflowExecutionDecisionAttributes;
 import com.uber.cadence.Decision;
-import com.uber.cadence.DecisionTaskCompletedEventAttributes;
 import com.uber.cadence.DecisionType;
 import com.uber.cadence.EventType;
 import com.uber.cadence.ExternalWorkflowExecutionCancelRequestedEventAttributes;
@@ -55,7 +54,6 @@
 import com.uber.cadence.internal.replay.HistoryHelper.DecisionEvents;
 import com.uber.cadence.internal.worker.WorkflowExecutionException;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.LinkedHashMap;
@@ -101,10 +99,6 @@ class DecisionsHelper {
   // TODO: removal of completed activities
   private final Map<String, Long> activityIdToScheduledEventId = new HashMap<>();
 
-  private byte[] workflowContextData;
-
-  private byte[] workfowContextFromLastDecisionCompletion;
-
   DecisionsHelper(PollForDecisionTaskResponse task) {
     this.task = task;
   }
@@ -583,25 +577,6 @@ public String toString() {
     return WorkflowExecutionUtils.prettyPrintDecisions(getDecisions());
   }
 
-  void setWorkflowContextData(byte[] workflowState) {
-    this.workflowContextData = workflowState;
-  }
-
-  /** @return new workflow state or null if it didn't change since the last decision completion */
-  byte[] getWorkflowContextDataToReturn() {
-    if (workfowContextFromLastDecisionCompletion == null
-        || !Arrays.equals(workfowContextFromLastDecisionCompletion, workflowContextData)) {
-      return workflowContextData;
-    }
-    return null;
-  }
-
-  void handleDecisionCompletion(
-      DecisionTaskCompletedEventAttributes decisionTaskCompletedEventAttributes) {
-    workfowContextFromLastDecisionCompletion =
-        decisionTaskCompletedEventAttributes.getExecutionContext();
-  }
-
   PollForDecisionTaskResponse getTask() {
     return task;
   }

src/main/java/com/uber/cadence/internal/replay/ReplayDecider.java

@@ -17,6 +17,7 @@
 
 package com.uber.cadence.internal.replay;
 
+import com.uber.cadence.Decision;
 import com.uber.cadence.EventType;
 import com.uber.cadence.GetWorkflowExecutionHistoryRequest;
 import com.uber.cadence.GetWorkflowExecutionHistoryResponse;
@@ -39,17 +40,17 @@
 import com.uber.cadence.workflow.Functions;
 import com.uber.m3.tally.Scope;
 import com.uber.m3.tally.Stopwatch;
-import org.apache.thrift.TException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import java.time.Duration;
 import java.util.Iterator;
+import java.util.List;
 import java.util.Objects;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Implements decider that relies on replay of a worklfow code. An instance of this class is created
@@ -155,7 +156,7 @@ private void processEvent(HistoryEvent event) throws Throwable {
         context.handleChildWorkflowExecutionTimedOut(event);
         break;
       case DecisionTaskCompleted:
-        handleDecisionTaskCompleted(event);
+        // NOOP
         break;
       case DecisionTaskScheduled:
         // NOOP
@@ -361,13 +362,10 @@ private void handleWorkflowExecutionSignaled(HistoryEvent event) {
         signalAttributes.getSignalName(), signalAttributes.getInput(), event.getEventId());
   }
 
-  private void handleDecisionTaskCompleted(HistoryEvent event) {
-    decisionsHelper.handleDecisionCompletion(event.getDecisionTaskCompletedEventAttributes());
-  }
-
   @Override
-  public void decide(PollForDecisionTaskResponse decisionTask) throws Throwable {
+  public List<Decision> decide(PollForDecisionTaskResponse decisionTask) throws Throwable {
     decideImpl(decisionTask, null);
+    return getDecisionsHelper().getDecisions();
   }
 
   private void decideImpl(PollForDecisionTaskResponse decisionTask, Functions.Proc query)

src/main/java/com/uber/cadence/internal/replay/ReplayDecisionTaskHandler.java

@@ -120,10 +120,7 @@ private Result processDecision(PollForDecisionTaskResponse decisionTask) throws
             ? createDecider(decisionTask)
             : cache.getOrCreate(decisionTask, this::createDecider);
     try {
-      decider.decide(decisionTask);
-      DecisionsHelper decisionsHelper = ((ReplayDecider) decider).getDecisionsHelper();
-      List<Decision> decisions = decisionsHelper.getDecisions();
-
+      List<Decision> decisions = decider.decide(decisionTask);
       if (log.isTraceEnabled()) {
         WorkflowExecution execution = decisionTask.getWorkflowExecution();
         log.trace(
@@ -148,7 +145,7 @@ private Result processDecision(PollForDecisionTaskResponse decisionTask) throws
                 + decisions.size()
                 + " new decisions");
       }
-      return createCompletedRequest(decisionTask, decisionsHelper, decisions);
+      return createCompletedRequest(decisionTask, decisions);
     } catch (Exception e) {
       if (stickyTaskListName != null) {
         cache.invalidate(decisionTask);
@@ -184,15 +181,11 @@ private Result processQuery(PollForDecisionTaskResponse decisionTask) {
   }
 
   private Result createCompletedRequest(
-      PollForDecisionTaskResponse decisionTask,
-      DecisionsHelper decisionsHelper,
-      List<Decision> decisions) {
-    byte[] context = decisionsHelper.getWorkflowContextDataToReturn();
+      PollForDecisionTaskResponse decisionTask, List<Decision> decisions) {
     RespondDecisionTaskCompletedRequest completedRequest =
         new RespondDecisionTaskCompletedRequest();
     completedRequest.setTaskToken(decisionTask.getTaskToken());
     completedRequest.setDecisions(decisions);
-    completedRequest.setExecutionContext(context);
 
     if (stickyTaskListName != null) {
       StickyExecutionAttributes attributes = new StickyExecutionAttributes();

src/main/java/com/uber/cadence/internal/testservice/TestWorkflowMutableStateImpl.java

@@ -82,10 +82,6 @@
 import com.uber.cadence.internal.testservice.StateMachines.TimerData;
 import com.uber.cadence.internal.testservice.StateMachines.WorkflowData;
 import com.uber.cadence.internal.testservice.TestWorkflowStore.TaskListId;
-import org.apache.thrift.TException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.DataInputStream;
@@ -106,6 +102,9 @@
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.LongSupplier;
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 class TestWorkflowMutableStateImpl implements TestWorkflowMutableState {
 

src/main/java/com/uber/cadence/internal/testservice/TestWorkflowStoreImpl.java

@@ -34,9 +34,6 @@
 import com.uber.cadence.WorkflowExecutionInfo;
 import com.uber.cadence.internal.common.WorkflowExecutionUtils;
 import com.uber.cadence.internal.testservice.RequestContext.Timer;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -49,6 +46,8 @@
 import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 class TestWorkflowStoreImpl implements TestWorkflowStore {
 

src/test/java/com/uber/cadence/internal/sync/DeterministicRunnerTest.java

@@ -27,6 +27,7 @@
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
+import com.uber.cadence.Decision;
 import com.uber.cadence.PollForDecisionTaskResponse;
 import com.uber.cadence.WorkflowQuery;
 import com.uber.cadence.WorkflowType;
@@ -790,7 +791,9 @@ private static class DetermisiticRunnerContainerDecider implements Decider {
     }
 
     @Override
-    public void decide(PollForDecisionTaskResponse decisionTask) throws Throwable {}
+    public List<Decision> decide(PollForDecisionTaskResponse decisionTask) throws Throwable {
+      return new ArrayList<>();
+    }
 
     @Override
     public byte[] query(PollForDecisionTaskResponse decisionTask, WorkflowQuery query)