implement getVersion (#166)

* implement getVersion

* review comments

* support the case where getVersion call is removed during event replay

* use data converter instead of lambda for encoding

* update class name

* turn off debug in test

* fix getversions get removed before another version marker decision

* address review comments

* add comments on addDecision, addAllMissingVersionMarker and nextDecisionEventId

* fix getOptionalDecisionEvent

Author: meiliang86

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

@@ -97,13 +97,6 @@ Consumer<Exception> scheduleActivityTask(
     attributes.setScheduleToStartTimeoutSeconds(
         (int) parameters.getScheduleToStartTimeoutSeconds());
     attributes.setStartToCloseTimeoutSeconds((int) parameters.getStartToCloseTimeoutSeconds());
-    //
-    // attributes.setTaskPriority(InternalUtils.taskPriorityToString(parameters.getTaskPriority()));
-    String activityId = parameters.getActivityId();
-    if (activityId == null) {
-      activityId = String.valueOf(decisions.getNextId());
-    }
-    attributes.setActivityId(activityId);
 
     String taskList = parameters.getTaskList();
     if (taskList != null && !taskList.isEmpty()) {
@@ -144,6 +137,11 @@ void handleActivityTaskCompleted(HistoryEvent event) {
         byte[] result = attributes.getResult();
         BiConsumer<byte[], Exception> completionHandle = scheduled.getCompletionCallback();
         completionHandle.accept(result, null);
+      } else {
+        throw new NonDeterminisicWorkflowError(
+            "Trying to complete activity event "
+                + attributes.getScheduledEventId()
+                + " that is not in scheduledActivities");
       }
     }
   }

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

@@ -17,13 +17,14 @@
 
 package com.uber.cadence.internal.replay;
 
-import com.uber.cadence.EventType;
 import com.uber.cadence.HistoryEvent;
 import com.uber.cadence.MarkerRecordedEventAttributes;
 import com.uber.cadence.StartTimerDecisionAttributes;
 import com.uber.cadence.TimerCanceledEventAttributes;
 import com.uber.cadence.TimerFiredEventAttributes;
-import com.uber.cadence.internal.replay.DecisionContext.MutableSideEffectData;
+import com.uber.cadence.converter.DataConverter;
+import com.uber.cadence.internal.replay.MarkerHandler.MarkerData;
+import com.uber.cadence.internal.sync.WorkflowInternal;
 import com.uber.cadence.workflow.Functions.Func;
 import com.uber.cadence.workflow.Functions.Func1;
 import java.util.HashMap;
@@ -33,6 +34,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
+import java.util.function.Predicate;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -41,6 +43,7 @@ final class ClockDecisionContext {
 
   private static final String SIDE_EFFECT_MARKER_NAME = "SideEffect";
   private static final String MUTABLE_SIDE_EFFECT_MARKER_NAME = "MutableSideEffect";
+  public static final String VERSION_MARKER_NAME = "Version";
 
   private static final Logger log = LoggerFactory.getLogger(ReplayDecider.class);
 
@@ -58,31 +61,6 @@ public void accept(Exception reason) {
     }
   }
 
-  private static final class MutableSideEffectResult {
-
-    private final byte[] data;
-
-    /**
-     * Count of how many times the mutableSideEffect was called since the last marker recorded. It
-     * is used to ensure that an updated value is returned after the same exact number of times
-     * during a replay.
-     */
-    private int accessCount;
-
-    private MutableSideEffectResult(byte[] data) {
-      this.data = data;
-    }
-
-    public byte[] getData() {
-      accessCount++;
-      return data;
-    }
-
-    public int getAccessCount() {
-      return accessCount;
-    }
-  }
-
   private final DecisionsHelper decisions;
 
   // key is startedEventId
@@ -94,11 +72,15 @@ public int getAccessCount() {
 
   // Key is side effect marker eventId
   private final Map<Long, byte[]> sideEffectResults = new HashMap<>();
-  // Key is mutableSideEffect id
-  private final Map<String, MutableSideEffectResult> mutableSideEffectResults = new HashMap<>();
+
+  private final MarkerHandler mutableSideEffectHandler;
+  private final MarkerHandler versionHandler;
 
   ClockDecisionContext(DecisionsHelper decisions) {
     this.decisions = decisions;
+    mutableSideEffectHandler =
+        new MarkerHandler(decisions, MUTABLE_SIDE_EFFECT_MARKER_NAME, () -> replaying);
+    versionHandler = new MarkerHandler(decisions, VERSION_MARKER_NAME, () -> replaying);
   }
 
   long currentTimeMillis() {
@@ -125,8 +107,7 @@ Consumer<Exception> createTimer(long delaySeconds, Consumer<Exception> callback)
     final OpenRequestInfo<?, Long> context = new OpenRequestInfo<>(firingTime);
     final StartTimerDecisionAttributes timer = new StartTimerDecisionAttributes();
     timer.setStartToFireTimeoutSeconds(delaySeconds);
-    timer.setTimerId(String.valueOf(decisions.getNextId()));
-    long startEventId = decisions.startTimer(timer, null);
+    long startEventId = decisions.startTimer(timer);
     context.setCompletionHandle((ctx, e) -> callback.accept(e));
     scheduledTimers.put(startEventId, context);
     return new TimerCancellationHandler(startEventId);
@@ -167,6 +148,7 @@ private void timerCancelled(long startEventId, Exception reason) {
   }
 
   byte[] sideEffect(Func<byte[]> func) {
+    decisions.addAllMissingVersionMarker(false, Optional.empty());
     long sideEffectEventId = decisions.getNextDecisionEventId();
     byte[] result;
     if (replaying) {
@@ -194,81 +176,54 @@ byte[] sideEffect(Func<byte[]> func) {
    * @return the latest value returned by func
    */
   Optional<byte[]> mutableSideEffect(
-      String id,
-      Func1<MutableSideEffectData, byte[]> markerDataSerializer,
-      Func1<byte[], MutableSideEffectData> markerDataDeserializer,
-      Func1<Optional<byte[]>, Optional<byte[]>> func) {
-    MutableSideEffectResult result = mutableSideEffectResults.get(id);
-    Optional<byte[]> stored;
-    if (result == null) {
-      stored = Optional.empty();
-    } else {
-      stored = Optional.of(result.getData());
-    }
-    long eventId = decisions.getNextDecisionEventId();
-    int accessCount = result == null ? 0 : result.getAccessCount();
-    if (replaying) {
-
-      Optional<byte[]> data =
-          getSideEffectDataFromHistory(eventId, id, accessCount, markerDataDeserializer);
-      if (data.isPresent()) {
-        // Need to insert marker to ensure that eventId is incremented
-        recordMutableSideEffectMarker(id, eventId, data.get(), accessCount, markerDataSerializer);
-        return data;
-      }
-      return stored;
-    }
-    Optional<byte[]> toStore = func.apply(stored);
-    if (toStore.isPresent()) {
-      byte[] data = toStore.get();
-      recordMutableSideEffectMarker(id, eventId, data, accessCount, markerDataSerializer);
-      return toStore;
-    }
-    return stored;
-  }
-
-  private Optional<byte[]> getSideEffectDataFromHistory(
-      long eventId,
-      String mutableSideEffectId,
-      int expectedAcccessCount,
-      Func1<byte[], MutableSideEffectData> markerDataDeserializer) {
-    HistoryEvent event = decisions.getDecisionEvent(eventId);
-    if (event.getEventType() != EventType.MarkerRecorded) {
-      return Optional.empty();
-    }
-    MarkerRecordedEventAttributes attributes = event.getMarkerRecordedEventAttributes();
-    String name = attributes.getMarkerName();
-    if (!MUTABLE_SIDE_EFFECT_MARKER_NAME.equals(name)) {
-      return Optional.empty();
-    }
-    MutableSideEffectData markerData = markerDataDeserializer.apply(attributes.getDetails());
-    // access count is used to not return data from the marker before the recorded number of calls
-    if (!mutableSideEffectId.equals(markerData.getId())
-        || markerData.getAccessCount() > expectedAcccessCount) {
-      return Optional.empty();
-    }
-    return Optional.of(markerData.getData());
-  }
-
-  private void recordMutableSideEffectMarker(
-      String id,
-      long eventId,
-      byte[] data,
-      int accessCount,
-      Func1<MutableSideEffectData, byte[]> markerDataSerializer) {
-    MutableSideEffectData dataObject = new MutableSideEffectData(id, eventId, data, accessCount);
-    byte[] details = markerDataSerializer.apply(dataObject);
-    mutableSideEffectResults.put(id, new MutableSideEffectResult(data));
-    decisions.recordMarker(MUTABLE_SIDE_EFFECT_MARKER_NAME, details);
+      String id, DataConverter converter, Func1<Optional<byte[]>, Optional<byte[]>> func) {
+    decisions.addAllMissingVersionMarker(false, Optional.empty());
+    return mutableSideEffectHandler.handle(id, converter, func);
   }
 
   void handleMarkerRecorded(HistoryEvent event) {
     MarkerRecordedEventAttributes attributes = event.getMarkerRecordedEventAttributes();
     String name = attributes.getMarkerName();
     if (SIDE_EFFECT_MARKER_NAME.equals(name)) {
       sideEffectResults.put(event.getEventId(), attributes.getDetails());
-    } else if (!MUTABLE_SIDE_EFFECT_MARKER_NAME.equals(name)) {
+    } else if (!MUTABLE_SIDE_EFFECT_MARKER_NAME.equals(name) && !VERSION_MARKER_NAME.equals(name)) {
       log.warn("Unexpected marker: " + event);
     }
   }
+
+  int getVersion(String changeId, DataConverter converter, int minSupported, int maxSupported) {
+    Predicate<byte[]> changeIdEquals =
+        (bytesInEvent) -> {
+          MarkerData markerData = converter.fromData(bytesInEvent, MarkerData.class);
+          return markerData.getId().equals(changeId);
+        };
+    decisions.addAllMissingVersionMarker(true, Optional.of(changeIdEquals));
+
+    Optional<byte[]> result =
+        versionHandler.handle(
+            changeId,
+            converter,
+            (stored) -> {
+              if (stored.isPresent()) {
+                return Optional.empty();
+              }
+              return Optional.of(converter.toData(maxSupported));
+            });
+
+    if (!result.isPresent()) {
+      return WorkflowInternal.DEFAULT_VERSION;
+    }
+    int version = converter.fromData(result.get(), Integer.class);
+    validateVersion(changeId, version, minSupported, maxSupported);
+    return version;
+  }
+
+  private void validateVersion(String changeID, int version, int minSupported, int maxSupported) {
+    if (version < minSupported || version > maxSupported) {
+      throw new Error(
+          String.format(
+              "Version %d of changeID %s is not supported. Supported version is between %d and %d.",
+              version, changeID, minSupported, maxSupported));
+    }
+  }
 }

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

@@ -19,6 +19,7 @@
 
 import com.uber.cadence.WorkflowExecution;
 import com.uber.cadence.WorkflowType;
+import com.uber.cadence.converter.DataConverter;
 import com.uber.cadence.workflow.Functions.Func;
 import com.uber.cadence.workflow.Functions.Func1;
 import com.uber.cadence.workflow.Promise;
@@ -34,37 +35,6 @@
  */
 public interface DecisionContext extends ReplayAware {
 
-  final class MutableSideEffectData {
-
-    private final String id;
-    private final long eventId;
-    private final byte[] data;
-    private final int accessCount;
-
-    public MutableSideEffectData(String id, long eventId, byte[] data, int accessCount) {
-      this.id = id;
-      this.eventId = eventId;
-      this.data = data;
-      this.accessCount = accessCount;
-    }
-
-    public String getId() {
-      return id;
-    }
-
-    public long getEventId() {
-      return eventId;
-    }
-
-    public byte[] getData() {
-      return data;
-    }
-
-    public int getAccessCount() {
-      return accessCount;
-    }
-  }
-
   WorkflowExecution getWorkflowExecution();
 
   // TODO: Add to Cadence
@@ -127,14 +97,8 @@ Consumer<Exception> signalWorkflowExecution(
 
   void continueAsNewOnCompletion(ContinueAsNewWorkflowExecutionParameters parameters);
 
-  /** Deterministic unique Id generator */
-  String generateUniqueId();
-
   Optional<byte[]> mutableSideEffect(
-      String id,
-      Func1<MutableSideEffectData, byte[]> markerDataSerializer,
-      Func1<byte[], MutableSideEffectData> markerDataDeserializer,
-      Func1<Optional<byte[]>, Optional<byte[]>> func);
+      String id, DataConverter dataConverter, Func1<Optional<byte[]>, Optional<byte[]>> func);
 
   /**
    * @return time of the {@link com.uber.cadence.PollForDecisionTaskResponse} start event of the
@@ -166,6 +130,23 @@ Optional<byte[]> mutableSideEffect(
    */
   byte[] sideEffect(Func<byte[]> func);
 
+  /**
+   * GetVersion is used to safely perform backwards incompatible changes to workflow definitions. It
+   * is not allowed to update workflow code while there are workflows running as it is going to
+   * break determinism. The solution is to have both old code that is used to replay existing
+   * workflows as well as the new one that is used when it is executed for the first time.
+   * GetVersion returns maxSupported version when executed for the first time. This version is
+   * recorded into the workflow history as a marker event. Even if maxSupported version is changed
+   * the version that was recorded is returned on replay. DefaultVersion constant contains version
+   * of code that wasn't versioned before.
+   *
+   * @param changeID identifier of a particular change
+   * @param minSupported min version supported for the change
+   * @param maxSupported max version supported for the change
+   * @return version
+   */
+  int getVersion(String changeID, DataConverter dataConverter, int minSupported, int maxSupported);
+
   /** @return scope to be used for metrics reporting. */
   Scope getMetricsScope();
 

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

@@ -23,6 +23,7 @@
 import com.uber.cadence.WorkflowExecution;
 import com.uber.cadence.WorkflowExecutionStartedEventAttributes;
 import com.uber.cadence.WorkflowType;
+import com.uber.cadence.converter.DataConverter;
 import com.uber.cadence.internal.metrics.ReplayAwareScope;
 import com.uber.cadence.workflow.Functions.Func;
 import com.uber.cadence.workflow.Functions.Func1;
@@ -172,11 +173,6 @@ public void continueAsNewOnCompletion(ContinueAsNewWorkflowExecutionParameters p
     workflowClient.continueAsNewOnCompletion(parameters);
   }
 
-  @Override
-  public String generateUniqueId() {
-    return workflowClient.generateUniqueId();
-  }
-
   void setReplayCurrentTimeMilliseconds(long replayCurrentTimeMilliseconds) {
     if (replayCurrentTimeMilliseconds < workflowClock.currentTimeMillis()) {
       throw new IllegalArgumentException("workflow clock moved back");
@@ -201,11 +197,14 @@ public byte[] sideEffect(Func<byte[]> func) {
 
   @Override
   public Optional<byte[]> mutableSideEffect(
-      String id,
-      Func1<MutableSideEffectData, byte[]> markerDataSerializer,
-      Func1<byte[], MutableSideEffectData> markerDataDeserializer,
-      Func1<Optional<byte[]>, Optional<byte[]>> func) {
-    return workflowClock.mutableSideEffect(id, markerDataSerializer, markerDataDeserializer, func);
+      String id, DataConverter converter, Func1<Optional<byte[]>, Optional<byte[]>> func) {
+    return workflowClock.mutableSideEffect(id, converter, func);
+  }
+
+  @Override
+  public int getVersion(
+      String changeID, DataConverter converter, int minSupported, int maxSupported) {
+    return workflowClock.getVersion(changeID, converter, minSupported, maxSupported);
   }
 
   @Override