Reset Replay Decider State (#211)

Author: halakaraki

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

@@ -66,7 +66,7 @@ public void accept(Exception reason) {
   // key is startedEventId
   private final Map<Long, OpenRequestInfo<?, Long>> scheduledTimers = new HashMap<>();
 
-  private long replayCurrentTimeMilliseconds;
+  private long replayCurrentTimeMilliseconds = -1;
 
   private boolean replaying = true;
 

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

@@ -28,12 +28,15 @@
 import com.uber.m3.tally.Scope;
 import java.util.Objects;
 import java.util.UUID;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public final class DeciderCache {
   private final String evictionEntryId = UUID.randomUUID().toString();
   private final int maxCacheSize;
   private final Scope metricsScope;
   private LoadingCache<String, WeightedCacheEntry<Decider>> cache;
+  private static final Logger log = LoggerFactory.getLogger(DeciderCache.class);
 
   public DeciderCache(int maxCacheSize, Scope scope) {
     Preconditions.checkArgument(maxCacheSize > 0, "Max cache size must be greater than 0");
@@ -91,8 +94,7 @@ public void evictNext() {
     int remainingSpace = (int) (maxCacheSize - cache.size());
     // Force eviction to happen
     cache.put(evictionEntryId, new WeightedCacheEntry<>(null, remainingSpace + 1));
-    cache.invalidate(evictionEntryId);
-    metricsScope.counter(MetricsType.STICKY_CACHE_TOTAL_FORCED_EVICTION).inc(1);
+    invalidate(evictionEntryId);
   }
 
   public void invalidate(PollForDecisionTaskResponse decisionTask) {
@@ -101,6 +103,7 @@ public void invalidate(PollForDecisionTaskResponse decisionTask) {
   }
 
   public void invalidate(String runId) {
+    metricsScope.counter(MetricsType.STICKY_CACHE_TOTAL_FORCED_EVICTION).inc(1);
     cache.invalidate(runId);
   }
 

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

@@ -192,6 +192,10 @@ void setReplayCurrentTimeMilliseconds(long replayCurrentTimeMilliseconds) {
     workflowClock.setReplayCurrentTimeMilliseconds(replayCurrentTimeMilliseconds);
   }
 
+  long getReplayCurrentTimeMilliseconds() {
+    return workflowClock.currentTimeMillis();
+  }
+
   @Override
   public boolean isReplaying() {
     return workflowClock.isReplaying();

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

@@ -20,9 +20,13 @@
 import com.uber.cadence.HistoryEvent;
 import java.util.ArrayList;
 import java.util.List;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 abstract class DecisionStateMachineBase implements DecisionStateMachine {
 
+  private static final Logger log = LoggerFactory.getLogger(DecisionStateMachineBase.class);
+
   protected DecisionState state = DecisionState.CREATED;
 
   protected List<String> stateHistory = new ArrayList<String>();

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

@@ -181,9 +181,13 @@ public void remove() {
   static class DecisionEventsIterator implements Iterator<DecisionEvents> {
 
     private EventsIterator events;
+    private long replayCurrentTimeMilliseconds;
 
-    DecisionEventsIterator(DecisionTaskWithHistoryIterator decisionTaskWithHistoryIterator) {
+    DecisionEventsIterator(
+        DecisionTaskWithHistoryIterator decisionTaskWithHistoryIterator,
+        long replayCurrentTimeMilliseconds) {
       this.events = new EventsIterator(decisionTaskWithHistoryIterator.getHistory());
+      this.replayCurrentTimeMilliseconds = replayCurrentTimeMilliseconds;
     }
 
     @Override
@@ -196,12 +200,17 @@ public DecisionEvents next() {
       List<HistoryEvent> decisionEvents = new ArrayList<>();
       List<HistoryEvent> newEvents = new ArrayList<>();
       boolean replay = true;
-      long replayCurrentTimeMilliseconds = -1;
       long nextDecisionEventId = -1;
       while (events.hasNext()) {
         HistoryEvent event = events.next();
         EventType eventType = event.getEventType();
-        // hasNext is for queries that do not have DecisionTaskStarted at the end of the history.
+
+        // Sticky workers receive an event history that starts with DecisionTaskCompleted
+        if (eventType == EventType.DecisionTaskCompleted && nextDecisionEventId == -1) {
+          nextDecisionEventId = event.getEventId() + 1;
+          break;
+        }
+
         if (eventType == EventType.DecisionTaskStarted || !events.hasNext()) {
           replayCurrentTimeMilliseconds = TimeUnit.NANOSECONDS.toMillis(event.getTimestamp());
           if (!events.hasNext()) {
@@ -216,10 +225,14 @@ public DecisionEvents next() {
             continue;
           } else if (peekedType == EventType.DecisionTaskCompleted) {
             events.next(); // consume DecisionTaskCompleted
-            nextDecisionEventId = peeked.getEventId() + 1; // +1 for next
+            nextDecisionEventId = peeked.getEventId() + 1; // +1 for next and skip over completed
             break;
           } else {
-            throw new Error("Unexpected event after DecisionTaskStarted: " + peeked);
+            throw new Error(
+                "Unexpected event after DecisionTaskStarted: "
+                    + peeked
+                    + " DecisionTaskStarted Event: "
+                    + event);
           }
         }
         newEvents.add(event);
@@ -245,9 +258,9 @@ public DecisionEvents next() {
   private final DecisionTaskWithHistoryIterator decisionTaskWithHistoryIterator;
   private final DecisionEventsIterator iterator;
 
-  HistoryHelper(DecisionTaskWithHistoryIterator decisionTasks) {
+  HistoryHelper(DecisionTaskWithHistoryIterator decisionTasks, long replayCurrentTimeMilliseconds) {
     this.decisionTaskWithHistoryIterator = decisionTasks;
-    this.iterator = new DecisionEventsIterator(decisionTasks);
+    this.iterator = new DecisionEventsIterator(decisionTasks, replayCurrentTimeMilliseconds);
   }
 
   public DecisionEventsIterator getIterator() {

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

@@ -374,7 +374,9 @@ private void decideImpl(PollForDecisionTaskResponse decisionTask, Functions.Proc
       DecisionTaskWithHistoryIterator decisionTaskWithHistoryIterator =
           new DecisionTaskWithHistoryIteratorImpl(
               decisionTask, Duration.ofSeconds(startedEvent.getTaskStartToCloseTimeoutSeconds()));
-      HistoryHelper historyHelper = new HistoryHelper(decisionTaskWithHistoryIterator);
+      HistoryHelper historyHelper =
+          new HistoryHelper(
+              decisionTaskWithHistoryIterator, context.getReplayCurrentTimeMilliseconds());
       DecisionEventsIterator iterator = historyHelper.getIterator();
       if ((decisionsHelper.getNextDecisionEventId()
               != historyHelper.getPreviousStartedEventId()
@@ -412,6 +414,8 @@ private void decideImpl(PollForDecisionTaskResponse decisionTask, Functions.Proc
         for (HistoryEvent event : decision.getDecisionEvents()) {
           processEvent(event);
         }
+        // Reset state to before running the event loop
+        decisionsHelper.handleDecisionTaskStartedEvent(decision);
       }
     } catch (Error e) {
       metricsScope.counter(MetricsType.DECISION_TASK_ERROR_COUNTER).inc(1);

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

@@ -54,27 +54,27 @@ public SyncWorkflowWorker(
       Duration stickyDecisionScheduleToStartTimeout,
       ThreadPoolExecutor workflowThreadPool) {
     Objects.requireNonNull(workflowThreadPool);
+    this.options = options;
 
     factory =
         new POJOWorkflowImplementationFactory(
             options.getDataConverter(),
             workflowThreadPool,
             interceptorFactory,
-            options.getMetricsScope(),
+            this.options.getMetricsScope(),
             cache);
 
     DecisionTaskHandler taskHandler =
         new ReplayDecisionTaskHandler(
             domain,
             factory,
             cache,
-            options,
+            this.options,
             stickyTaskListName,
             stickyDecisionScheduleToStartTimeout,
             service);
 
-    worker = new WorkflowWorker(service, domain, taskList, options, taskHandler);
-    this.options = options;
+    worker = new WorkflowWorker(service, domain, taskList, this.options, taskHandler);
   }
 
   public void setWorkflowImplementationTypes(Class<?>[] workflowImplementationTypes) {

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

@@ -402,28 +402,25 @@ public Factory(IWorkflowService workflowService, String domain, FactoryOptions f
         return;
       }
 
-      factoryOptions.metricsScope.tagged(
-          new ImmutableMap.Builder<String, String>(2)
-              .put(MetricsTag.DOMAIN, domain)
-              .put(MetricsTag.TASK_LIST, getStickyTaskListName())
-              .build());
+      Scope metricsScope =
+          factoryOptions.metricsScope.tagged(
+              new ImmutableMap.Builder<String, String>(2)
+                  .put(MetricsTag.DOMAIN, domain)
+                  .put(MetricsTag.TASK_LIST, getHostName())
+                  .build());
 
-      this.cache = new DeciderCache(factoryOptions.cacheMaximumSize, factoryOptions.metricsScope);
+      this.cache = new DeciderCache(factoryOptions.cacheMaximumSize, metricsScope);
 
       dispatcher = new PollDecisionTaskDispatcherFactory(workflowService).create();
       stickyPoller =
           new Poller<>(
               id.toString(),
               new WorkflowPollTaskFactory(
-                      workflowService,
-                      domain,
-                      getStickyTaskListName(),
-                      factoryOptions.metricsScope,
-                      id.toString())
+                      workflowService, domain, getStickyTaskListName(), metricsScope, id.toString())
                   .get(),
               dispatcher,
               factoryOptions.stickyWorkflowPollerOptions,
-              factoryOptions.metricsScope);
+              metricsScope);
     }
 
     public Worker newWorker(String taskList) {
@@ -500,7 +497,8 @@ DeciderCache getCache() {
       return this.cache;
     }
 
-    private String getHostName() {
+    @VisibleForTesting
+    String getHostName() {
       try {
         return InetAddress.getLocalHost().getHostName();
       } catch (UnknownHostException e) {
@@ -514,18 +512,6 @@ private String getStickyTaskListName() {
           : null;
     }
 
-    private SingleWorkerOptions getDefaultSingleWorkerOptions() {
-      return Worker.toWorkflowOptions(new Builder().build(), domain, getStickyTaskListName());
-    }
-
-    private PollerOptions getDefaultPollerOptions(SingleWorkerOptions options) {
-      PollerOptions pollerOptions = options.getPollerOptions();
-      if (pollerOptions.getPollThreadNamePrefix() == null) {
-        pollerOptions = new PollerOptions.Builder(pollerOptions).build();
-      }
-      return pollerOptions;
-    }
-
     enum State {
       Initial,
       Started,