Child workflow server side retry (#231)

Author: mfateev

src/main/java/com/uber/cadence/client/WorkflowClient.java

@@ -103,7 +103,9 @@
 public interface WorkflowClient {
 
   /** Use this constant as a query type to get a workflow stack trace. */
-  String QUERY_TYPE_STACK_TRCE = "__stack_trace";
+  String QUERY_TYPE_STACK_TRACE = "__stack_trace";
+  /** Replays workflow to the current state and returns empty result or error if replay failed. */
+  String QUERY_TYPE_REPLAY_ONLY = "__replay_only";
 
   /**
    * Creates worker that connects to the local instance of the Cadence Service that listens on a

src/main/java/com/uber/cadence/common/WorkflowExecutionHistory.java

@@ -0,0 +1,120 @@
+/*
+ *  Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *  Modifications copyright (C) 2017 Uber Technologies, Inc.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License"). You may not
+ *  use this file except in compliance with the License. A copy of the License is
+ *  located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ *  or in the "license" file accompanying this file. This file is distributed on
+ *  an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ *  express or implied. See the License for the specific language governing
+ *  permissions and limitations under the License.
+ */
+
+package com.uber.cadence.common;
+
+import com.google.common.collect.ImmutableList;
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
+import com.google.gson.JsonDeserializationContext;
+import com.google.gson.JsonDeserializer;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonParseException;
+import com.google.gson.JsonPrimitive;
+import com.google.gson.JsonSerializationContext;
+import com.google.gson.JsonSerializer;
+import com.uber.cadence.EventType;
+import com.uber.cadence.HistoryEvent;
+import com.uber.cadence.WorkflowExecution;
+import java.lang.reflect.Type;
+import java.nio.ByteBuffer;
+import java.util.Base64;
+import java.util.List;
+
+/** Contains workflow execution ids and the history */
+public final class WorkflowExecutionHistory {
+  private final String workflowId;
+  private final String runId;
+  private final List<HistoryEvent> events;
+
+  public WorkflowExecutionHistory(String workflowId, String runId, List<HistoryEvent> events) {
+    this.workflowId = workflowId;
+    this.runId = runId;
+    checkHistory(events);
+    this.events = ImmutableList.copyOf(events);
+  }
+
+  public WorkflowExecutionHistory(WorkflowExecution workflowExecution, List<HistoryEvent> events) {
+    this.workflowId = workflowExecution.getWorkflowId();
+    this.runId = workflowExecution.getRunId();
+    checkHistory(events);
+    this.events = ImmutableList.copyOf(events);
+  }
+
+  public static WorkflowExecutionHistory fromJson(String serialized) {
+    GsonBuilder gsonBuilder = new GsonBuilder();
+    gsonBuilder.registerTypeAdapter(ByteBuffer.class, new ByteBufferJsonDeserializer());
+    Gson gson = gsonBuilder.create();
+    WorkflowExecutionHistory result = gson.fromJson(serialized, WorkflowExecutionHistory.class);
+    checkHistory(result.getEvents());
+    return result;
+  }
+
+  private static void checkHistory(List<HistoryEvent> events) {
+    if (events == null || events.size() == 0) {
+      throw new IllegalArgumentException("Empty history");
+    }
+    HistoryEvent startedEvent = events.get(0);
+    if (startedEvent.getEventType() != EventType.WorkflowExecutionStarted) {
+      throw new IllegalArgumentException(
+          "First event is not WorkflowExecutionStarted but " + startedEvent);
+    }
+    if (startedEvent.getWorkflowExecutionStartedEventAttributes() == null) {
+      throw new IllegalArgumentException("First event is corrupted");
+    }
+  }
+
+  public String toJson() {
+    GsonBuilder gsonBuilder = new GsonBuilder();
+    Gson gson = gsonBuilder.create();
+    return gson.toJson(this);
+  }
+
+  public String getWorkflowId() {
+    return workflowId;
+  }
+
+  public String getRunId() {
+    return runId;
+  }
+
+  public WorkflowExecution getWorkflowExecution() {
+    return new WorkflowExecution().setWorkflowId(workflowId).setRunId(runId);
+  }
+
+  public List<HistoryEvent> getEvents() {
+    return events;
+  }
+
+  private static final class ByteBufferJsonDeserializer
+      implements JsonDeserializer<ByteBuffer>, JsonSerializer<ByteBuffer> {
+
+    @Override
+    public JsonElement serialize(ByteBuffer value, Type type, JsonSerializationContext ctx) {
+      if (value.arrayOffset() > 0) {
+        throw new IllegalArgumentException("non zero value array offset: " + value.arrayOffset());
+      }
+      return new JsonPrimitive(Base64.getEncoder().encodeToString(value.array()));
+    }
+
+    @Override
+    public ByteBuffer deserialize(JsonElement e, Type type, JsonDeserializationContext ctx)
+        throws JsonParseException {
+      return ByteBuffer.wrap(Base64.getDecoder().decode(e.getAsString()));
+    }
+  }
+}

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

@@ -17,6 +17,9 @@
 
 package com.uber.cadence.internal.common;
 
+import static java.nio.charset.StandardCharsets.UTF_8;
+
+import com.google.common.io.CharStreams;
 import com.google.gson.Gson;
 import com.google.gson.GsonBuilder;
 import com.google.gson.JsonElement;
@@ -49,11 +52,15 @@
 import com.uber.cadence.client.WorkflowTerminatedException;
 import com.uber.cadence.client.WorkflowTimedOutException;
 import com.uber.cadence.common.RetryOptions;
+import com.uber.cadence.common.WorkflowExecutionHistory;
 import com.uber.cadence.serviceclient.IWorkflowService;
+import java.io.File;
+import java.io.IOException;
+import java.io.Reader;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.lang.reflect.Modifier;
-import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
 import java.time.Duration;
 import java.util.Collection;
 import java.util.Date;
@@ -141,7 +148,7 @@ private static byte[] getResultFromCloseEvent(
         return closeEvent.getWorkflowExecutionCompletedEventAttributes().getResult();
       case WorkflowExecutionCanceled:
         byte[] details = closeEvent.getWorkflowExecutionCanceledEventAttributes().getDetails();
-        String message = details != null ? new String(details, StandardCharsets.UTF_8) : null;
+        String message = details != null ? new String(details, UTF_8) : null;
         throw new CancellationException(message);
       case WorkflowExecutionFailed:
         WorkflowExecutionFailedEventAttributes failed =
@@ -754,7 +761,7 @@ private static String prettyPrintObject(
       return (String) object;
     }
     if (clz.equals(byte[].class)) {
-      return new String((byte[]) object, StandardCharsets.UTF_8);
+      return new String((byte[]) object, UTF_8);
     }
 
     if (clz.equals(Date.class)) {
@@ -833,7 +840,7 @@ private static String prettyPrintObject(
         result.append(" = ");
         // Pretty print JSON serialized exceptions.
         if (name.equals("getDetails") && value instanceof byte[]) {
-          String details = new String((byte[]) value, StandardCharsets.UTF_8);
+          String details = new String((byte[]) value, UTF_8);
           details = prettyPrintJson(details, INDENTATION + INDENTATION);
           // GSON pretty prints, but doesn't let to set an initial indentation.
           // Thus indenting the pretty printed JSON through regexp :(.
@@ -961,4 +968,19 @@ public static EventType getEventTypeForDecision(DecisionType decisionType) {
     }
     throw new IllegalArgumentException("Unknown decisionType");
   }
+
+  public static WorkflowExecutionHistory readHistoryFromResource(String resourceFileName)
+      throws IOException {
+    ClassLoader classLoader = WorkflowExecutionUtils.class.getClassLoader();
+    String historyUrl = classLoader.getResource(resourceFileName).getFile();
+    File historyFile = new File(historyUrl);
+    return readHistory(historyFile);
+  }
+
+  public static WorkflowExecutionHistory readHistory(File historyFile) throws IOException {
+    try (Reader reader = Files.newBufferedReader(historyFile.toPath(), UTF_8)) {
+      String jsonHistory = CharStreams.toString(reader);
+      return WorkflowExecutionHistory.fromJson(jsonHistory);
+    }
+  }
 }

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

@@ -95,6 +95,13 @@ Consumer<Exception> startChildWorkflow(
       Consumer<WorkflowExecution> executionCallback,
       BiConsumer<byte[], Exception> callback);
 
+  /**
+   * Is the next event in the history is child workflow initiated event and it has attached retry
+   * policy. Used for backwards compatibility with the code that used local workflow retry when
+   * RetryOptions were specified.
+   */
+  boolean isServerSideChildWorkflowRetry();
+
   Consumer<Exception> signalWorkflowExecution(
       SignalExternalWorkflowParameters signalParameters, BiConsumer<Void, Exception> callback);
 

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

@@ -167,6 +167,11 @@ public Consumer<Exception> startChildWorkflow(
     return workflowClient.startChildWorkflow(parameters, executionCallback, callback);
   }
 
+  @Override
+  public boolean isServerSideChildWorkflowRetry() {
+    return workflowClient.isChildWorkflowExecutionStartedWithRetryOptions();
+  }
+
   @Override
   public Consumer<Exception> signalWorkflowExecution(
       SignalExternalWorkflowParameters signalParameters, BiConsumer<Void, Exception> callback) {

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

@@ -44,6 +44,7 @@
 import com.uber.cadence.SignalExternalWorkflowExecutionDecisionAttributes;
 import com.uber.cadence.StartChildWorkflowExecutionDecisionAttributes;
 import com.uber.cadence.StartChildWorkflowExecutionFailedEventAttributes;
+import com.uber.cadence.StartChildWorkflowExecutionInitiatedEventAttributes;
 import com.uber.cadence.StartTimerDecisionAttributes;
 import com.uber.cadence.TaskList;
 import com.uber.cadence.TimerCanceledEventAttributes;
@@ -197,6 +198,27 @@ long startChildWorkflowExecution(StartChildWorkflowExecutionDecisionAttributes c
     return nextDecisionEventId;
   }
 
+  /**
+   * @return true if it is not replay or retryOptions are present in the
+   *     StartChildWorkflowExecutionInitiated event.
+   */
+  boolean isChildWorkflowExecutionStartedWithRetryOptions() {
+    Optional<HistoryEvent> optionalEvent = getOptionalDecisionEvent(nextDecisionEventId);
+    if (!optionalEvent.isPresent()) {
+      return true;
+    }
+    HistoryEvent event = optionalEvent.get();
+    if (event.getEventType() != EventType.StartChildWorkflowExecutionInitiated) {
+      return false;
+    }
+    StartChildWorkflowExecutionInitiatedEventAttributes attr =
+        event.getStartChildWorkflowExecutionInitiatedEventAttributes();
+    if (attr == null) {
+      throw new Error("Corrupted event: " + event);
+    }
+    return attr.getRetryPolicy() != null;
+  }
+
   void handleStartChildWorkflowExecutionInitiated(HistoryEvent event) {
     DecisionStateMachine decision =
         getDecision(new DecisionId(DecisionTarget.CHILD_WORKFLOW, event.getEventId()));

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

@@ -20,7 +20,9 @@
 import com.uber.cadence.ChildPolicy;
 import com.uber.cadence.WorkflowIdReusePolicy;
 import com.uber.cadence.WorkflowType;
+import com.uber.cadence.internal.common.RetryParameters;
 import java.util.Arrays;
+import java.util.Objects;
 
 public final class StartChildWorkflowExecutionParameters {
 
@@ -46,6 +48,8 @@ public static final class Builder {
 
     private WorkflowIdReusePolicy workflowIdReusePolicy;
 
+    private RetryParameters retryParameters;
+
     public Builder setDomain(String domain) {
       this.domain = domain;
       return this;
@@ -97,6 +101,11 @@ public Builder setWorkflowIdReusePolicy(WorkflowIdReusePolicy workflowIdReusePol
       return this;
     }
 
+    public Builder setRetryParameters(RetryParameters retryParameters) {
+      this.retryParameters = retryParameters;
+      return this;
+    }
+
     public StartChildWorkflowExecutionParameters build() {
       return new StartChildWorkflowExecutionParameters(
           domain,
@@ -108,7 +117,8 @@ public StartChildWorkflowExecutionParameters build() {
           workflowId,
           workflowType,
           childPolicy,
-          workflowIdReusePolicy);
+          workflowIdReusePolicy,
+          retryParameters);
     }
   }
 
@@ -132,6 +142,8 @@ public StartChildWorkflowExecutionParameters build() {
 
   private final WorkflowIdReusePolicy workflowIdReusePolicy;
 
+  private final RetryParameters retryParameters;
+
   private StartChildWorkflowExecutionParameters(
       String domain,
       byte[] input,
@@ -142,7 +154,8 @@ private StartChildWorkflowExecutionParameters(
       String workflowId,
       WorkflowType workflowType,
       ChildPolicy childPolicy,
-      WorkflowIdReusePolicy workflowIdReusePolicy) {
+      WorkflowIdReusePolicy workflowIdReusePolicy,
+      RetryParameters retryParameters) {
     this.domain = domain;
     this.input = input;
     this.control = control;
@@ -153,6 +166,7 @@ private StartChildWorkflowExecutionParameters(
     this.workflowType = workflowType;
     this.childPolicy = childPolicy;
     this.workflowIdReusePolicy = workflowIdReusePolicy;
+    this.retryParameters = retryParameters;
   }
 
   public String getDomain() {
@@ -195,6 +209,46 @@ public WorkflowIdReusePolicy getWorkflowIdReusePolicy() {
     return workflowIdReusePolicy;
   }
 
+  public RetryParameters getRetryParameters() {
+    return retryParameters;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) return true;
+    if (o == null || getClass() != o.getClass()) return false;
+    StartChildWorkflowExecutionParameters that = (StartChildWorkflowExecutionParameters) o;
+    return executionStartToCloseTimeoutSeconds == that.executionStartToCloseTimeoutSeconds
+        && taskStartToCloseTimeoutSeconds == that.taskStartToCloseTimeoutSeconds
+        && Objects.equals(domain, that.domain)
+        && Objects.equals(control, that.control)
+        && Arrays.equals(input, that.input)
+        && Objects.equals(taskList, that.taskList)
+        && Objects.equals(workflowId, that.workflowId)
+        && Objects.equals(workflowType, that.workflowType)
+        && childPolicy == that.childPolicy
+        && workflowIdReusePolicy == that.workflowIdReusePolicy
+        && Objects.equals(retryParameters, that.retryParameters);
+  }
+
+  @Override
+  public int hashCode() {
+    int result =
+        Objects.hash(
+            domain,
+            control,
+            executionStartToCloseTimeoutSeconds,
+            taskList,
+            taskStartToCloseTimeoutSeconds,
+            workflowId,
+            workflowType,
+            childPolicy,
+            workflowIdReusePolicy,
+            retryParameters);
+    result = 31 * result + Arrays.hashCode(input);
+    return result;
+  }
+
   @Override
   public String toString() {
     return "StartChildWorkflowExecutionParameters{"
@@ -222,6 +276,8 @@ public String toString() {
         + childPolicy
         + ", workflowIdReusePolicy="
         + workflowIdReusePolicy
+        + ", retryParameters="
+        + retryParameters
         + '}';
   }
 }