Add server side cron schedule support (#248)

* Add cron parser library and plumb through cronSchedule in workflow options

* Add cronSchedule in start workflow params and last result in sync decision context

* Add test framework change and unit test for cron schedule

* Fix execution timer delay calculation in test framework

* Implement getLasCompletionResult in test framework

* Add java docs

* Address review comments

* Fix unit test

Author: meiliang86

build.gradle

@@ -51,6 +51,7 @@ dependencies {
     compile group: 'com.google.code.gson', name: 'gson', version: '2.8.5'
     compile group: 'com.uber.m3', name: 'tally-core', version: '0.2.3'
     compile group: 'com.google.guava', name: 'guava', version: '27.0.1-jre'
+    compile group: 'com.cronutils', name: 'cron-utils', version: '8.0.0'
     testCompile group: 'junit', name: 'junit', version: '4.12'
     testCompile group: 'com.googlecode.junit-toolbox', name: 'junit-toolbox', version: '2.4'
     testCompile group: 'ch.qos.logback', name: 'logback-classic', version: '1.2.3'

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

@@ -19,8 +19,15 @@
 
 import static com.uber.cadence.internal.common.OptionsUtils.roundUpToSeconds;
 
+import com.cronutils.model.Cron;
+import com.cronutils.model.CronType;
+import com.cronutils.model.definition.CronDefinition;
+import com.cronutils.model.definition.CronDefinitionBuilder;
+import com.cronutils.parser.CronParser;
+import com.google.common.base.Strings;
 import com.uber.cadence.ChildPolicy;
 import com.uber.cadence.WorkflowIdReusePolicy;
+import com.uber.cadence.common.CronSchedule;
 import com.uber.cadence.common.MethodRetry;
 import com.uber.cadence.common.RetryOptions;
 import com.uber.cadence.internal.common.OptionsUtils;
@@ -31,13 +38,14 @@
 public final class WorkflowOptions {
 
   public static WorkflowOptions merge(
-      WorkflowMethod a, MethodRetry methodRetry, WorkflowOptions o) {
+      WorkflowMethod a, MethodRetry methodRetry, CronSchedule cronSchedule, WorkflowOptions o) {
     if (a == null) {
       return new WorkflowOptions.Builder(o).validateBuildWithDefaults();
     }
     if (o == null) {
       o = new WorkflowOptions.Builder().build();
     }
+    String cronAnnotation = cronSchedule == null ? "" : cronSchedule.value();
     return new WorkflowOptions.Builder()
         .setWorkflowIdReusePolicy(
             OptionsUtils.merge(
@@ -53,6 +61,7 @@ public static WorkflowOptions merge(
         .setTaskList(OptionsUtils.merge(a.taskList(), o.getTaskList(), String.class))
         .setChildPolicy(o.getChildPolicy())
         .setRetryOptions(RetryOptions.merge(methodRetry, o.getRetryOptions()))
+        .setCronSchedule(OptionsUtils.merge(cronAnnotation, o.getCronSchedule(), String.class))
         .validateBuildWithDefaults();
   }
 
@@ -72,6 +81,8 @@ public static final class Builder {
 
     private RetryOptions retryOptions;
 
+    private String cronSchedule;
+
     public Builder() {}
 
     public Builder(WorkflowOptions o) {
@@ -85,6 +96,7 @@ public Builder(WorkflowOptions o) {
       this.taskList = o.taskList;
       this.childPolicy = o.childPolicy;
       this.retryOptions = o.retryOptions;
+      this.cronSchedule = o.cronSchedule;
     }
 
     /**
@@ -164,6 +176,11 @@ public Builder setRetryOptions(RetryOptions retryOptions) {
       return this;
     }
 
+    public Builder setCronSchedule(String cronSchedule) {
+      this.cronSchedule = cronSchedule;
+      return this;
+    }
+
     public WorkflowOptions build() {
       return new WorkflowOptions(
           workflowId,
@@ -172,7 +189,8 @@ public WorkflowOptions build() {
           taskStartToCloseTimeout,
           taskList,
           childPolicy,
-          retryOptions);
+          retryOptions,
+          cronSchedule);
     }
 
     /**
@@ -200,6 +218,14 @@ public WorkflowOptions validateBuildWithDefaults() {
               "RetryOptions must specify either expiration or maximum attempts");
         }
       }
+
+      if (!Strings.isNullOrEmpty(cronSchedule)) {
+        CronDefinition cronDefinition = CronDefinitionBuilder.instanceDefinitionFor(CronType.UNIX);
+        CronParser parser = new CronParser(cronDefinition);
+        Cron cron = parser.parse(cronSchedule);
+        cron.validate();
+      }
+
       return new WorkflowOptions(
           workflowId,
           policy,
@@ -208,7 +234,8 @@ public WorkflowOptions validateBuildWithDefaults() {
               taskStartToCloseTimeout, OptionsUtils.DEFAULT_TASK_START_TO_CLOSE_TIMEOUT),
           taskList,
           childPolicy,
-          retryOptions);
+          retryOptions,
+          cronSchedule);
     }
   }
 
@@ -226,21 +253,25 @@ public WorkflowOptions validateBuildWithDefaults() {
 
   private RetryOptions retryOptions;
 
+  private String cronSchedule;
+
   private WorkflowOptions(
       String workflowId,
       WorkflowIdReusePolicy workflowIdReusePolicy,
       Duration executionStartToCloseTimeout,
       Duration taskStartToCloseTimeout,
       String taskList,
       ChildPolicy childPolicy,
-      RetryOptions retryOptions) {
+      RetryOptions retryOptions,
+      String cronSchedule) {
     this.workflowId = workflowId;
     this.workflowIdReusePolicy = workflowIdReusePolicy;
     this.executionStartToCloseTimeout = executionStartToCloseTimeout;
     this.taskStartToCloseTimeout = taskStartToCloseTimeout;
     this.taskList = taskList;
     this.childPolicy = childPolicy;
     this.retryOptions = retryOptions;
+    this.cronSchedule = cronSchedule;
   }
 
   public String getWorkflowId() {
@@ -271,6 +302,10 @@ public RetryOptions getRetryOptions() {
     return retryOptions;
   }
 
+  public String getCronSchedule() {
+    return cronSchedule;
+  }
+
   @Override
   public boolean equals(Object o) {
     if (this == o) return true;
@@ -282,7 +317,8 @@ public boolean equals(Object o) {
         && Objects.equals(taskStartToCloseTimeout, that.taskStartToCloseTimeout)
         && Objects.equals(taskList, that.taskList)
         && childPolicy == that.childPolicy
-        && Objects.equals(retryOptions, that.retryOptions);
+        && Objects.equals(retryOptions, that.retryOptions)
+        && Objects.equals(cronSchedule, that.cronSchedule);
   }
 
   @Override
@@ -294,7 +330,8 @@ public int hashCode() {
         taskStartToCloseTimeout,
         taskList,
         childPolicy,
-        retryOptions);
+        retryOptions,
+        cronSchedule);
   }
 
   @Override
@@ -316,6 +353,9 @@ public String toString() {
         + childPolicy
         + ", retryOptions="
         + retryOptions
+        + ", cronSchedule='"
+        + cronSchedule
+        + '\''
         + '}';
   }
 }

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

@@ -0,0 +1,38 @@
+/*
+ *  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 java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * CronSchedule - Optional cron schedule for workflow. If a cron schedule is specified, the workflow
+ * will run as a cron based on the schedule. The scheduling will be based on UTC time. Schedule for
+ * next run only happen after the current run is completed/failed/timeout. If a RetryPolicy is also
+ * supplied, and the workflow failed or timeout, the workflow will be retried based on the retry
+ * policy. While the workflow is retrying, it won't schedule its next run. If next schedule is due
+ * while workflow is running (or retrying), then it will skip that schedule. Cron workflow will not
+ * stop until it is terminated or cancelled.
+ */
+@Retention(RetentionPolicy.RUNTIME)
+@Target(ElementType.METHOD)
+public @interface CronSchedule {
+  String value();
+}

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

@@ -48,6 +48,8 @@ public final class StartWorkflowExecutionParameters {
 
   private RetryParameters retryParameters;
 
+  private String cronSchedule;
+
   /**
    * Returns the value of the WorkflowId property for this object.
    *
@@ -282,13 +284,20 @@ public void setRetryParameters(RetryParameters retryParameters) {
     this.retryParameters = retryParameters;
   }
 
+  public String getCronSchedule() {
+    return cronSchedule;
+  }
+
+  public void setCronSchedule(String cronSchedule) {
+    this.cronSchedule = cronSchedule;
+  }
+
   public StartWorkflowExecutionParameters withRetryParameters(RetryParameters retryParameters) {
     this.retryParameters = retryParameters;
     return this;
   }
 
-  public static StartWorkflowExecutionParameters createStartWorkflowExecutionParametersFromOptions(
-      WorkflowOptions options) {
+  public static StartWorkflowExecutionParameters fromWorkflowOptions(WorkflowOptions options) {
     StartWorkflowExecutionParameters parameters = new StartWorkflowExecutionParameters();
     parameters.setExecutionStartToCloseTimeoutSeconds(
         getSeconds(options.getExecutionStartToCloseTimeout()));
@@ -314,6 +323,10 @@ public static StartWorkflowExecutionParameters createStartWorkflowExecutionParam
       }
       parameters.setRetryParameters(rp);
     }
+
+    if (options.getCronSchedule() != null) {
+      parameters.setCronSchedule(options.getCronSchedule());
+    }
     return parameters;
   }
 
@@ -347,6 +360,9 @@ public String toString() {
         + workflowIdReusePolicy
         + ", retryParameters="
         + retryParameters
+        + ", cronSchedule='"
+        + cronSchedule
+        + '\''
         + '}';
   }
 
@@ -363,7 +379,8 @@ public boolean equals(Object o) {
         && Arrays.equals(input, that.input)
         && childPolicy == that.childPolicy
         && workflowIdReusePolicy == that.workflowIdReusePolicy
-        && Objects.equals(retryParameters, that.retryParameters);
+        && Objects.equals(retryParameters, that.retryParameters)
+        && Objects.equals(cronSchedule, that.cronSchedule);
   }
 
   @Override
@@ -377,21 +394,9 @@ public int hashCode() {
             taskStartToCloseTimeoutSeconds,
             childPolicy,
             workflowIdReusePolicy,
-            retryParameters);
+            retryParameters,
+            cronSchedule);
     result = 31 * result + Arrays.hashCode(input);
     return result;
   }
-
-  public StartWorkflowExecutionParameters copy() {
-    StartWorkflowExecutionParameters result = new StartWorkflowExecutionParameters();
-    result.setInput(input);
-    result.setExecutionStartToCloseTimeoutSeconds(executionStartToCloseTimeoutSeconds);
-    result.setTaskStartToCloseTimeoutSeconds(taskStartToCloseTimeoutSeconds);
-    result.setTaskList(taskList);
-    result.setWorkflowId(workflowId);
-    result.setWorkflowType(workflowType);
-    result.setChildPolicy(childPolicy);
-    result.setRetryParameters(retryParameters.copy());
-    return result;
-  }
 }

src/main/java/com/uber/cadence/internal/external/GenericWorkflowClientExternalImpl.java

@@ -17,6 +17,7 @@
 
 package com.uber.cadence.internal.external;
 
+import com.google.common.base.Strings;
 import com.uber.cadence.*;
 import com.uber.cadence.internal.common.CheckedExceptionWrapper;
 import com.uber.cadence.internal.common.RetryParameters;
@@ -109,6 +110,9 @@ private WorkflowExecution startWorkflowInternal(StartWorkflowExecutionParameters
               .setNonRetriableErrorReasons(retryParameters.getNonRetriableErrorReasons());
       request.setRetryPolicy(retryPolicy);
     }
+    if (!Strings.isNullOrEmpty(startParameters.getCronSchedule())) {
+      request.setCronSchedule(startParameters.getCronSchedule());
+    }
 
     //        if(startParameters.getChildPolicy() != null) {
     //            request.setChildPolicy(startParameters.getChildPolicy());

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

@@ -175,7 +175,7 @@ public Thread newThread(Runnable r) {
 
   private static SyncDecisionContext newDummySyncDecisionContext() {
     return new SyncDecisionContext(
-        new DummyDecisionContext(), JsonDataConverter.getInstance(), (next) -> next);
+        new DummyDecisionContext(), JsonDataConverter.getInstance(), (next) -> next, null);
   }
 
   SyncDecisionContext getDecisionContext() {

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

@@ -79,11 +79,13 @@ final class SyncDecisionContext implements WorkflowInterceptor {
   private final WorkflowInterceptor headInterceptor;
   private final WorkflowTimers timers = new WorkflowTimers();
   private final Map<String, Functions.Func1<byte[], byte[]>> queryCallbacks = new HashMap<>();
+  private final byte[] lastCompletionResult;
 
   public SyncDecisionContext(
       DecisionContext context,
       DataConverter converter,
-      Function<WorkflowInterceptor, WorkflowInterceptor> interceptorFactory) {
+      Function<WorkflowInterceptor, WorkflowInterceptor> interceptorFactory,
+      byte[] lastCompletionResult) {
     this.context = context;
     this.converter = converter;
     WorkflowInterceptor interceptor = interceptorFactory.apply(this);
@@ -92,6 +94,7 @@ public SyncDecisionContext(
       interceptor = this;
     }
     this.headInterceptor = interceptor;
+    this.lastCompletionResult = lastCompletionResult;
   }
 
   /**
@@ -593,4 +596,13 @@ public Scope getMetricsScope() {
   public boolean isLoggingEnabledInReplay() {
     return context.getEnableLoggingInReplay();
   }
+
+  public <R> R getLastCompletionResult(Class<R> resultClass, Type resultType) {
+    if (lastCompletionResult == null || lastCompletionResult.length == 0) {
+      return null;
+    }
+
+    DataConverter dataConverter = getDataConverter();
+    return dataConverter.fromData(lastCompletionResult, resultClass, resultType);
+  }
 }

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

@@ -78,13 +78,18 @@ public void start(HistoryEvent event, DecisionContext context) {
     if (workflow == null) {
       throw new IllegalArgumentException("Unknown workflow type: " + workflowType);
     }
-    SyncDecisionContext syncContext =
-        new SyncDecisionContext(context, dataConverter, interceptorFactory);
     if (event.getEventType() != EventType.WorkflowExecutionStarted) {
       throw new IllegalArgumentException(
           "first event is not WorkflowExecutionStarted, but " + event.getEventType());
     }
 
+    SyncDecisionContext syncContext =
+        new SyncDecisionContext(
+            context,
+            dataConverter,
+            interceptorFactory,
+            event.getWorkflowExecutionStartedEventAttributes().getLastCompletionResult());
+
     workflowProc =
         new WorkflowRunnable(
             syncContext, workflow, event.getWorkflowExecutionStartedEventAttributes());

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

@@ -359,4 +359,8 @@ public static Logger getLogger(String name) {
     return new ReplayAwareLogger(
         logger, WorkflowInternal::isReplaying, WorkflowInternal::isLoggingEnabledInReplay);
   }
+
+  public static <R> R getLastCompletionResult(Class<R> resultClass, Type resultType) {
+    return getRootDecisionContext().getLastCompletionResult(resultClass, resultType);
+  }
 }