Change markers to use headers to serialize internal fields (#324)

* Change local activity marker to use headers
* Change mutable side effects and version marker to use header
* Update pretty print object method for test framework
* Add backward compatibility test

Author: meiliang86

src/main/java/com/uber/cadence/internal/common/LocalActivityMarkerData.java

@@ -17,18 +17,25 @@
 
 package com.uber.cadence.internal.common;
 
+import com.google.common.base.Strings;
 import com.uber.cadence.ActivityType;
+import com.uber.cadence.Header;
+import com.uber.cadence.MarkerRecordedEventAttributes;
 import com.uber.cadence.RespondActivityTaskCanceledRequest;
 import com.uber.cadence.RespondActivityTaskFailedRequest;
+import com.uber.cadence.converter.DataConverter;
+import com.uber.m3.util.ImmutableMap;
+import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 
 public final class LocalActivityMarkerData {
+  private static final String LOCAL_ACTIVITY_HEADER_KEY = "LocalActivityHeader";
+
   public static final class Builder {
     private String activityId;
     private String activityType;
     private String errReason;
-    private byte[] errJson;
     private byte[] result;
     private long replayTimeMillis;
     private int attempt;
@@ -47,13 +54,13 @@ public Builder setActivityType(ActivityType activityType) {
 
     public Builder setTaskFailedRequest(RespondActivityTaskFailedRequest request) {
       this.errReason = request.getReason();
-      this.errJson = request.getDetails();
+      this.result = request.getDetails();
       return this;
     }
 
     public Builder setTaskCancelledRequest(RespondActivityTaskCanceledRequest request) {
       this.errReason = new String(request.getDetails(), StandardCharsets.UTF_8);
-      this.errJson = request.getDetails();
+      this.result = request.getDetails();
       this.isCancelled = true;
       return this;
     }
@@ -85,77 +92,114 @@ public LocalActivityMarkerData build() {
           replayTimeMillis,
           result,
           errReason,
-          errJson,
           attempt,
           backoff,
           isCancelled);
     }
   }
 
-  private final String activityId;
-  private final String activityType;
-  private final String errReason;
-  private final byte[] errJson;
+  private static class LocalActivityMarkerHeader {
+    private final String activityId;
+    private final String activityType;
+    private final String errReason;
+    private final long replayTimeMillis;
+    private final int attempt;
+    private final Duration backoff;
+    private final boolean isCancelled;
+
+    LocalActivityMarkerHeader(
+        String activityId,
+        String activityType,
+        long replayTimeMillis,
+        String errReason,
+        int attempt,
+        Duration backoff,
+        boolean isCancelled) {
+      this.activityId = activityId;
+      this.activityType = activityType;
+      this.replayTimeMillis = replayTimeMillis;
+      this.errReason = errReason;
+      this.attempt = attempt;
+      this.backoff = backoff;
+      this.isCancelled = isCancelled;
+    }
+  }
+
+  private final LocalActivityMarkerHeader headers;
   private final byte[] result;
-  private final long replayTimeMillis;
-  private final int attempt;
-  private final Duration backoff;
-  private final boolean isCancelled;
 
   private LocalActivityMarkerData(
       String activityId,
       String activityType,
       long replayTimeMillis,
       byte[] result,
       String errReason,
-      byte[] errJson,
       int attempt,
       Duration backoff,
       boolean isCancelled) {
-    this.activityId = activityId;
-    this.activityType = activityType;
-    this.replayTimeMillis = replayTimeMillis;
+    this.headers =
+        new LocalActivityMarkerHeader(
+            activityId, activityType, replayTimeMillis, errReason, attempt, backoff, isCancelled);
     this.result = result;
-    this.errReason = errReason;
-    this.errJson = errJson;
-    this.attempt = attempt;
-    this.backoff = backoff;
-    this.isCancelled = isCancelled;
+  }
+
+  private LocalActivityMarkerData(LocalActivityMarkerHeader headers, byte[] result) {
+    this.headers = headers;
+    this.result = result;
+    if (headers == null) {
+      System.out.println("test");
+    }
   }
 
   public String getActivityId() {
-    return activityId;
+    return headers.activityId;
   }
 
   public String getActivityType() {
-    return activityType;
+    return headers.activityType;
   }
 
   public String getErrReason() {
-    return errReason;
+    return headers.errReason;
   }
 
   public byte[] getErrJson() {
-    return errJson;
+    return Strings.isNullOrEmpty(headers.errReason) ? null : result;
   }
 
   public byte[] getResult() {
     return result;
   }
 
   public long getReplayTimeMillis() {
-    return replayTimeMillis;
+    return headers.replayTimeMillis;
   }
 
   public int getAttempt() {
-    return attempt;
+    return headers.attempt;
   }
 
   public Duration getBackoff() {
-    return backoff;
+    return headers.backoff;
   }
 
   public boolean getIsCancelled() {
-    return isCancelled;
+    return headers.isCancelled;
+  }
+
+  public Header getHeader(DataConverter converter) {
+    byte[] headerData = converter.toData(headers);
+    Header header = new Header();
+    header.setFields(ImmutableMap.of(LOCAL_ACTIVITY_HEADER_KEY, ByteBuffer.wrap(headerData)));
+    return header;
+  }
+
+  public static LocalActivityMarkerData fromEventAttributes(
+      MarkerRecordedEventAttributes attributes, DataConverter converter) {
+    ByteBuffer byteBuffer = attributes.getHeader().getFields().get(LOCAL_ACTIVITY_HEADER_KEY);
+    byte[] bytes = org.apache.thrift.TBaseHelper.byteBufferToByteArray(byteBuffer);
+    LocalActivityMarkerHeader header =
+        converter.fromData(bytes, LocalActivityMarkerHeader.class, LocalActivityMarkerHeader.class);
+    return new LocalActivityMarkerData(header, attributes.getDetails());
   }
 }

src/main/java/com/uber/cadence/internal/common/WorkflowExecutionUtils.java

@@ -39,7 +39,6 @@
 import com.uber.cadence.History;
 import com.uber.cadence.HistoryEvent;
 import com.uber.cadence.HistoryEventFilterType;
-import com.uber.cadence.MarkerRecordedEventAttributes;
 import com.uber.cadence.StartWorkflowExecutionRequest;
 import com.uber.cadence.TaskList;
 import com.uber.cadence.WorkflowExecution;
@@ -61,7 +60,7 @@
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.lang.reflect.Modifier;
-import java.nio.charset.Charset;
+import java.nio.ByteBuffer;
 import java.nio.file.Files;
 import java.time.Duration;
 import java.util.Collection;
@@ -713,27 +712,9 @@ private static String prettyPrintHistoryEvent(HistoryEvent event, long firstTime
       result.append(String.format(" [%s ms]", timestamp));
     }
     result.append(" ");
-
-    if (event.getEventType() == EventType.MarkerRecorded) {
-      MarkerRecordedEventAttributes markerAttributes = event.getMarkerRecordedEventAttributes();
-      result
-          .append("{\n")
-          .append("    MarkerName = ")
-          .append(markerAttributes.getMarkerName())
-          .append(";\n");
-      result
-          .append("    DecisionTaskCompletedEventId = ")
-          .append(markerAttributes.getDecisionTaskCompletedEventId())
-          .append(";\n");
-      result
-          .append("    Details = ")
-          .append(new String(markerAttributes.getDetails(), Charset.defaultCharset()))
-          .append(";\n  }");
-    } else {
-      result.append(
-          prettyPrintObject(
-              getEventAttributes(event), "getFieldValue", true, INDENTATION, false, false));
-    }
+    result.append(
+        prettyPrintObject(
+            getEventAttributes(event), "getFieldValue", true, INDENTATION, false, false));
 
     return result.toString();
   }
@@ -784,7 +765,9 @@ private static String prettyPrintObject(
     if (clz.equals(byte[].class)) {
       return new String((byte[]) object, UTF_8);
     }
-
+    if (ByteBuffer.class.isAssignableFrom(clz)) {
+      return new String(((ByteBuffer) object).array(), UTF_8);
+    }
     if (clz.equals(Date.class)) {
       return String.valueOf(object);
     }
@@ -797,9 +780,40 @@ private static String prettyPrintObject(
     if (clz.equals(WorkflowType.class)) {
       return String.valueOf(((WorkflowType) object).getName());
     }
-
+    if (Map.Entry.class.isAssignableFrom(clz)) {
+      result.append(
+          prettyPrintObject(
+              ((Map.Entry) object).getKey(),
+              methodToSkip,
+              skipNullsAndEmptyCollections,
+              "",
+              skipLevel,
+              printTypeName));
+      result.append("=");
+      result.append(
+          prettyPrintObject(
+              ((Map.Entry) object).getValue(),
+              methodToSkip,
+              skipNullsAndEmptyCollections,
+              "",
+              skipLevel,
+              printTypeName));
+      return result.toString();
+    }
     if (Map.class.isAssignableFrom(clz)) {
-      return String.valueOf(object);
+      result.append("{ ");
+
+      String prefix = "";
+      for (Object entry : ((Map) object).entrySet()) {
+        result.append(prefix);
+        prefix = ", ";
+        result.append(
+            prettyPrintObject(
+                entry, methodToSkip, skipNullsAndEmptyCollections, "", skipLevel, printTypeName));
+      }
+
+      result.append(" }");
+      return result.toString();
     }
     if (Collection.class.isAssignableFrom(clz)) {
       return String.valueOf(object);

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

@@ -25,9 +25,7 @@
 import com.uber.cadence.TimerCanceledEventAttributes;
 import com.uber.cadence.TimerFiredEventAttributes;
 import com.uber.cadence.converter.DataConverter;
-import com.uber.cadence.converter.JsonDataConverter;
 import com.uber.cadence.internal.common.LocalActivityMarkerData;
-import com.uber.cadence.internal.replay.MarkerHandler.MarkerData;
 import com.uber.cadence.internal.sync.WorkflowInternal;
 import com.uber.cadence.internal.worker.LocalActivityWorker;
 import com.uber.cadence.workflow.ActivityFailureException;
@@ -88,20 +86,23 @@ public void accept(Exception reason) {
   private final Map<String, OpenRequestInfo<byte[], ActivityType>> pendingLaTasks = new HashMap<>();
   private final Map<String, ExecuteLocalActivityParameters> unstartedLaTasks = new HashMap<>();
   private final ReplayDecider replayDecider;
+  private final DataConverter dataConverter;
   private final Lock laTaskLock = new ReentrantLock();
   private final Condition taskCondition = laTaskLock.newCondition();
   private boolean taskCompleted = false;
 
   ClockDecisionContext(
       DecisionsHelper decisions,
       BiFunction<LocalActivityWorker.Task, Duration, Boolean> laTaskPoller,
-      ReplayDecider replayDecider) {
+      ReplayDecider replayDecider,
+      DataConverter dataConverter) {
     this.decisions = decisions;
     mutableSideEffectHandler =
         new MarkerHandler(decisions, MUTABLE_SIDE_EFFECT_MARKER_NAME, () -> replaying);
     versionHandler = new MarkerHandler(decisions, VERSION_MARKER_NAME, () -> replaying);
     this.laTaskPoller = laTaskPoller;
     this.replayDecider = replayDecider;
+    this.dataConverter = dataConverter;
   }
 
   public long currentTimeMillis() {
@@ -192,7 +193,7 @@ byte[] sideEffect(Func<byte[]> func) {
         throw new Error("sideEffect function failed", e);
       }
     }
-    decisions.recordMarker(SIDE_EFFECT_MARKER_NAME, result);
+    decisions.recordMarker(SIDE_EFFECT_MARKER_NAME, null, result);
     return result;
   }
 
@@ -222,16 +223,13 @@ void handleMarkerRecorded(HistoryEvent event) {
 
   private void handleLocalActivityMarker(MarkerRecordedEventAttributes attributes) {
     LocalActivityMarkerData marker =
-        JsonDataConverter.getInstance()
-            .fromData(
-                attributes.getDetails(),
-                LocalActivityMarkerData.class,
-                LocalActivityMarkerData.class);
+        LocalActivityMarkerData.fromEventAttributes(attributes, dataConverter);
 
     if (pendingLaTasks.containsKey(marker.getActivityId())) {
       log.debug("Handle LocalActivityMarker for activity " + marker.getActivityId());
 
-      decisions.recordMarker(LOCAL_ACTIVITY_MARKER_NAME, attributes.getDetails());
+      decisions.recordMarker(
+          LOCAL_ACTIVITY_MARKER_NAME, marker.getHeader(dataConverter), attributes.getDetails());
 
       OpenRequestInfo<byte[], ActivityType> scheduled =
           pendingLaTasks.remove(marker.getActivityId());
@@ -242,8 +240,7 @@ private void handleLocalActivityMarker(MarkerRecordedEventAttributes attributes)
         failure = new CancellationException(marker.getErrReason());
       } else if (marker.getErrJson() != null) {
         Throwable cause =
-            JsonDataConverter.getInstance()
-                .fromData(marker.getErrJson(), Throwable.class, Throwable.class);
+            dataConverter.fromData(marker.getErrJson(), Throwable.class, Throwable.class);
         ActivityType activityType = new ActivityType();
         activityType.setName(marker.getActivityType());
         failure =
@@ -271,10 +268,10 @@ private void handleLocalActivityMarker(MarkerRecordedEventAttributes attributes)
   }
 
   int getVersion(String changeId, DataConverter converter, int minSupported, int maxSupported) {
-    Predicate<byte[]> changeIdEquals =
-        (bytesInEvent) -> {
-          MarkerData markerData =
-              converter.fromData(bytesInEvent, MarkerData.class, MarkerData.class);
+    Predicate<MarkerRecordedEventAttributes> changeIdEquals =
+        (attributes) -> {
+          MarkerHandler.MarkerInterface markerData =
+              MarkerHandler.MarkerInterface.fromEventAttributes(attributes, converter);
           return markerData.getId().equals(changeId);
         };
     decisions.addAllMissingVersionMarker(true, Optional.of(changeIdEquals));