Added support for activity server side retry (#233)

* Added support for server side activity retry

* Added activity retry to the testservice

* Added Activity.getHeartbeatDetails()

* Changed RetryPolicy to not require expiration when maxAttempts is present

Author: mfateev

build.gradle

@@ -182,7 +182,7 @@ test {
         events "passed", "skipped", "failed"
         exceptionFormat "full"
         // Uncomment the following line if you want to see test logs in gradlew run.
-        // showStandardStreams true
+        showStandardStreams true
     }
 }
 

src/main/java/com/uber/cadence/activity/Activity.java

@@ -22,6 +22,8 @@
 import com.uber.cadence.serviceclient.IWorkflowService;
 import com.uber.cadence.workflow.ActivityException;
 import com.uber.cadence.workflow.ActivityTimeoutException;
+import java.lang.reflect.Type;
+import java.util.Optional;
 import java.util.concurrent.CancellationException;
 
 /**
@@ -238,6 +240,32 @@ public static void heartbeat(Object details) throws CancellationException {
     ActivityInternal.recordActivityHeartbeat(details);
   }
 
+  /**
+   * Extracts heartbeat details from the last failed attempt. This is used in combination with retry
+   * options. An activity could be scheduled with an optional {@link
+   * com.uber.cadence.common.RetryOptions} on {@link ActivityOptions}. If an activity failed then
+   * the server would attempt to dispatch another activity task to retry according to the retry
+   * options. If there was heartbeat details reported by the activity from the failed attempt, the
+   * details would be delivered along with the activity task for the retry attempt. The activity
+   * could extract the details by {@link #getHeartbeatDetails(Class)}() and resume from the
+   * progress.
+   *
+   * @param detailsClass type of the heartbeat details
+   */
+  public static <V> Optional<V> getHeartbeatDetails(Class<V> detailsClass) {
+    return ActivityInternal.getHeartbeatDetails(detailsClass, detailsClass);
+  }
+
+  /**
+   * Similar to {@link #getHeartbeatDetails(Class)}. Use when details is of a generic type.
+   *
+   * @param detailsClass type of the heartbeat details
+   * @param detailsType type including generic information of the heartbeat details.
+   */
+  public static <V> Optional<V> getHeartbeatDetails(Class<V> detailsClass, Type detailsType) {
+    return ActivityInternal.getHeartbeatDetails(detailsClass, detailsType);
+  }
+
   /**
    * @return an instance of the Simple Workflow Java client that is the same used by the invoked
    *     activity worker. It can be useful if activity wants to use WorkflowClient for some

src/main/java/com/uber/cadence/activity/ActivityOptions.java

@@ -175,13 +175,17 @@ public ActivityOptions validateAndBuildWithDefaults() {
       if (heartbeatTimeout == null) {
         heartbeat = scheduleToClose;
       }
+      RetryOptions ro = null;
+      if (retryOptions != null) {
+        ro = new RetryOptions.Builder(retryOptions).validateBuildWithDefaults();
+      }
       return new ActivityOptions(
           roundUpToSeconds(heartbeat),
           roundUpToSeconds(scheduleToClose),
           roundUpToSeconds(scheduleToStart),
           roundUpToSeconds(startToClose),
           taskList,
-          retryOptions);
+          ro);
     }
   }
 

src/main/java/com/uber/cadence/activity/ActivityTask.java

@@ -56,4 +56,6 @@ public interface ActivityTask {
   Duration getStartToCloseTimeout();
 
   Duration getHeartbeatTimeout();
+
+  byte[] getHeartbeatDetails();
 }

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

@@ -208,35 +208,15 @@ public RetryOptions build() {
 
     /** Validates property values and builds RetryOptions with default values. */
     public RetryOptions validateBuildWithDefaults() {
-      validate();
       double backoff = backoffCoefficient;
       if (backoff == 0d) {
         backoff = DEFAULT_BACKOFF_COEFFICIENT;
       }
-      return new RetryOptions(
-          initialInterval, backoff, expiration, maximumAttempts, maximumInterval, doNotRetry);
-    }
-
-    private void validate() {
-      if (initialInterval == null) {
-        throw new IllegalArgumentException("required property initialInterval not set");
-      }
-      if (expiration == null) {
-        throw new IllegalArgumentException("required property expiration is not set");
-      }
-      if (maximumInterval != null && maximumInterval.compareTo(initialInterval) == -1) {
-        throw new IllegalStateException(
-            "maximumInterval("
-                + maximumInterval
-                + ") cannot be smaller than initialInterval("
-                + initialInterval);
-      }
-      if (backoffCoefficient != 0d && backoffCoefficient < 1d) {
-        throw new IllegalArgumentException("coefficient less than 1: " + backoffCoefficient);
-      }
-      if (maximumAttempts != 0 && maximumAttempts < 0) {
-        throw new IllegalArgumentException("negative maximum attempts");
-      }
+      RetryOptions result =
+          new RetryOptions(
+              initialInterval, backoff, expiration, maximumAttempts, maximumInterval, doNotRetry);
+      result.validate();
+      return result;
     }
   }
 
@@ -291,6 +271,10 @@ public void validate() {
     if (initialInterval == null) {
       throw new IllegalStateException("required property initialInterval not set");
     }
+    if (expiration == null && maximumAttempts <= 0) {
+      throw new IllegalArgumentException(
+          "both MaximumAttempts and Expiration on retry policy are not set, at least one of them must be set");
+    }
     if (maximumInterval != null && maximumInterval.compareTo(initialInterval) == -1) {
       throw new IllegalStateException(
           "maximumInterval("

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

@@ -17,10 +17,15 @@
 
 package com.uber.cadence.internal.common;
 
+import static com.uber.cadence.internal.common.OptionsUtils.roundUpToSeconds;
+
+import com.uber.cadence.RetryPolicy;
+import com.uber.cadence.common.RetryOptions;
 import com.uber.m3.util.ImmutableList;
+import java.util.ArrayList;
 import java.util.List;
 
-public class RetryParameters {
+public final class RetryParameters {
 
   public int initialIntervalInSeconds;
   public double backoffCoefficient;
@@ -29,6 +34,29 @@ public class RetryParameters {
   public List<String> nonRetriableErrorReasons;
   public int expirationIntervalInSeconds;
 
+  public RetryParameters(RetryOptions retryOptions) {
+    setBackoffCoefficient(retryOptions.getBackoffCoefficient());
+    setExpirationIntervalInSeconds(
+        (int) roundUpToSeconds(retryOptions.getExpiration()).getSeconds());
+    setMaximumAttempts(retryOptions.getMaximumAttempts());
+    setInitialIntervalInSeconds(
+        (int) roundUpToSeconds(retryOptions.getInitialInterval()).getSeconds());
+    setMaximumIntervalInSeconds(
+        (int) roundUpToSeconds(retryOptions.getMaximumInterval()).getSeconds());
+    // Use exception type name as the reason
+    List<String> reasons = new ArrayList<>();
+    // Use exception type name as the reason
+    List<Class<? extends Throwable>> doNotRetry = retryOptions.getDoNotRetry();
+    if (doNotRetry != null) {
+      for (Class<? extends Throwable> r : doNotRetry) {
+        reasons.add(r.getName());
+      }
+      setNonRetriableErrorReasons(reasons);
+    }
+  }
+
+  public RetryParameters() {}
+
   public int getInitialIntervalInSeconds() {
     return initialIntervalInSeconds;
   }
@@ -87,4 +115,32 @@ public RetryParameters copy() {
     result.setBackoffCoefficient(backoffCoefficient);
     return result;
   }
+
+  public RetryPolicy toRetryPolicy() {
+    return new RetryPolicy()
+        .setNonRetriableErrorReasons(getNonRetriableErrorReasons())
+        .setMaximumAttempts(getMaximumAttempts())
+        .setInitialIntervalInSeconds(getInitialIntervalInSeconds())
+        .setExpirationIntervalInSeconds(getExpirationIntervalInSeconds())
+        .setBackoffCoefficient(getBackoffCoefficient())
+        .setMaximumIntervalInSeconds(getMaximumIntervalInSeconds());
+  }
+
+  @Override
+  public String toString() {
+    return "RetryParameters{"
+        + "initialIntervalInSeconds="
+        + initialIntervalInSeconds
+        + ", backoffCoefficient="
+        + backoffCoefficient
+        + ", maximumIntervalInSeconds="
+        + maximumIntervalInSeconds
+        + ", maximumAttempts="
+        + maximumAttempts
+        + ", nonRetriableErrorReasons="
+        + nonRetriableErrorReasons
+        + ", expirationIntervalInSeconds="
+        + expirationIntervalInSeconds
+        + '}';
+  }
 }

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

@@ -94,7 +94,8 @@ public class WorkflowExecutionUtils {
       new RetryOptions.Builder()
           .setBackoffCoefficient(2)
           .setInitialInterval(Duration.ofMillis(500))
-          .setMaximumInterval(Duration.ofSeconds(10))
+          .setMaximumInterval(Duration.ofSeconds(30))
+          .setMaximumAttempts(Integer.MAX_VALUE)
           .setDoNotRetry(BadRequestError.class, EntityNotExistsError.class)
           .build();
 

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

@@ -26,6 +26,7 @@
 import com.uber.cadence.ScheduleActivityTaskDecisionAttributes;
 import com.uber.cadence.TaskList;
 import com.uber.cadence.TimeoutType;
+import com.uber.cadence.internal.common.RetryParameters;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.CancellationException;
@@ -81,6 +82,10 @@ public void accept(Exception cause) {
     this.decisions = decisions;
   }
 
+  public boolean isActivityScheduledWithRetryOptions() {
+    return decisions.isActivityScheduledWithRetryOptions();
+  }
+
   Consumer<Exception> scheduleActivityTask(
       ExecuteActivityParameters parameters, BiConsumer<byte[], Exception> callback) {
     final OpenRequestInfo<byte[], ActivityType> context =
@@ -111,6 +116,11 @@ Consumer<Exception> scheduleActivityTask(
       tl.setName(taskList);
       attributes.setTaskList(tl);
     }
+    RetryParameters retryParameters = parameters.getRetryParameters();
+    if (retryParameters != null) {
+      attributes.setRetryPolicy(retryParameters.toRetryPolicy());
+    }
+
     long scheduledEventId = decisions.scheduleActivityTask(attributes);
     context.setCompletionHandle(callback);
     scheduledActivities.put(scheduledEventId, context);

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

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

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

@@ -172,6 +172,11 @@ public boolean isServerSideChildWorkflowRetry() {
     return workflowClient.isChildWorkflowExecutionStartedWithRetryOptions();
   }
 
+  @Override
+  public boolean isServerSideActivityRetry() {
+    return activityClient.isActivityScheduledWithRetryOptions();
+  }
+
   @Override
   public Consumer<Exception> signalWorkflowExecution(
       SignalExternalWorkflowParameters signalParameters, BiConsumer<Void, Exception> callback) {