Cron child workflow (#264)

* Add cron schedule to child workflow option
* Fix unit test
* Review comments

Author: meiliang86

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

@@ -50,6 +50,8 @@ public static final class Builder {
 
     private RetryParameters retryParameters;
 
+    private String cronSchedule;
+
     public Builder setDomain(String domain) {
       this.domain = domain;
       return this;
@@ -106,6 +108,11 @@ public Builder setRetryParameters(RetryParameters retryParameters) {
       return this;
     }
 
+    public Builder setCronSchedule(String cronSchedule) {
+      this.cronSchedule = cronSchedule;
+      return this;
+    }
+
     public StartChildWorkflowExecutionParameters build() {
       return new StartChildWorkflowExecutionParameters(
           domain,
@@ -118,7 +125,8 @@ public StartChildWorkflowExecutionParameters build() {
           workflowType,
           childPolicy,
           workflowIdReusePolicy,
-          retryParameters);
+          retryParameters,
+          cronSchedule);
     }
   }
 
@@ -144,6 +152,8 @@ public StartChildWorkflowExecutionParameters build() {
 
   private final RetryParameters retryParameters;
 
+  private final String cronSchedule;
+
   private StartChildWorkflowExecutionParameters(
       String domain,
       byte[] input,
@@ -155,7 +165,8 @@ private StartChildWorkflowExecutionParameters(
       WorkflowType workflowType,
       ChildPolicy childPolicy,
       WorkflowIdReusePolicy workflowIdReusePolicy,
-      RetryParameters retryParameters) {
+      RetryParameters retryParameters,
+      String cronSchedule) {
     this.domain = domain;
     this.input = input;
     this.control = control;
@@ -167,6 +178,7 @@ private StartChildWorkflowExecutionParameters(
     this.childPolicy = childPolicy;
     this.workflowIdReusePolicy = workflowIdReusePolicy;
     this.retryParameters = retryParameters;
+    this.cronSchedule = cronSchedule;
   }
 
   public String getDomain() {
@@ -213,6 +225,10 @@ public RetryParameters getRetryParameters() {
     return retryParameters;
   }
 
+  public String getCronSchedule() {
+    return cronSchedule;
+  }
+
   @Override
   public boolean equals(Object o) {
     if (this == o) return true;
@@ -228,7 +244,8 @@ public boolean equals(Object o) {
         && Objects.equals(workflowType, that.workflowType)
         && childPolicy == that.childPolicy
         && workflowIdReusePolicy == that.workflowIdReusePolicy
-        && Objects.equals(retryParameters, that.retryParameters);
+        && Objects.equals(retryParameters, that.retryParameters)
+        && Objects.equals(cronSchedule, that.cronSchedule);
   }
 
   @Override
@@ -244,7 +261,8 @@ public int hashCode() {
             workflowType,
             childPolicy,
             workflowIdReusePolicy,
-            retryParameters);
+            retryParameters,
+            cronSchedule);
     result = 31 * result + Arrays.hashCode(input);
     return result;
   }
@@ -278,6 +296,8 @@ public String toString() {
         + workflowIdReusePolicy
         + ", retryParameters="
         + retryParameters
+        + ", cronSchedule="
+        + cronSchedule
         + '}';
   }
 }

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

@@ -17,6 +17,7 @@
 
 package com.uber.cadence.internal.replay;
 
+import com.google.common.base.Strings;
 import com.uber.cadence.ChildPolicy;
 import com.uber.cadence.ChildWorkflowExecutionCanceledEventAttributes;
 import com.uber.cadence.ChildWorkflowExecutionCompletedEventAttributes;
@@ -147,6 +148,11 @@ Consumer<Exception> startChildWorkflow(
     if (retryParameters != null) {
       attributes.setRetryPolicy(retryParameters.toRetryPolicy());
     }
+
+    if (!Strings.isNullOrEmpty(parameters.getCronSchedule())) {
+      attributes.setCronSchedule(parameters.getCronSchedule());
+    }
+
     long initiatedEventId = decisions.startChildWorkflowExecution(attributes);
     final OpenChildWorkflowRequestInfo context =
         new OpenChildWorkflowRequestInfo(executionCallback);

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

@@ -21,6 +21,7 @@
 import static com.uber.cadence.internal.common.InternalUtils.getWorkflowMethod;
 import static com.uber.cadence.internal.common.InternalUtils.getWorkflowType;
 
+import com.uber.cadence.common.CronSchedule;
 import com.uber.cadence.common.MethodRetry;
 import com.uber.cadence.internal.common.InternalUtils;
 import com.uber.cadence.workflow.ChildWorkflowOptions;
@@ -45,9 +46,10 @@ class ChildWorkflowInvocationHandler implements InvocationHandler {
     WorkflowMethod workflowAnnotation = workflowMethod.getAnnotation(WorkflowMethod.class);
     String workflowType = getWorkflowType(workflowMethod, workflowAnnotation);
     MethodRetry retryAnnotation = workflowMethod.getAnnotation(MethodRetry.class);
+    CronSchedule cronSchedule = workflowMethod.getAnnotation(CronSchedule.class);
 
     ChildWorkflowOptions merged =
-        ChildWorkflowOptions.merge(workflowAnnotation, retryAnnotation, options);
+        ChildWorkflowOptions.merge(workflowAnnotation, retryAnnotation, cronSchedule, options);
     this.stub = new ChildWorkflowStubImpl(workflowType, merged, decisionContext);
   }
 

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

@@ -107,14 +107,6 @@ public void addActivityImplementation(Object activity) {
 
   private ActivityTaskHandler.Result mapToActivityFailure(
       String activityType, Throwable failure, Scope metricsScope) {
-    if (failure instanceof Error) {
-      Map<String, String> tags =
-          new ImmutableMap.Builder<String, String>(1)
-              .put(MetricsTag.ACTIVITY_TYPE, activityType)
-              .build();
-      metricsScope.tagged(tags).counter(MetricsType.ACTIVITY_TASK_ERROR_COUNTER).inc(1);
-      throw (Error) failure;
-    }
 
     if (failure instanceof ActivityCancelledException) {
       throw new CancellationException(failure.getMessage());
@@ -129,7 +121,16 @@ private ActivityTaskHandler.Result mapToActivityFailure(
               dataConverter.toData(timeoutException.getDetails()));
     }
 
-    metricsScope.counter(MetricsType.ACTIVITY_EXEC_FAILED_COUNTER).inc(1);
+    Map<String, String> activityTypeTag =
+        new ImmutableMap.Builder<String, String>(1)
+            .put(MetricsTag.ACTIVITY_TYPE, activityType)
+            .build();
+    if (failure instanceof Error) {
+      metricsScope.tagged(activityTypeTag).counter(MetricsType.ACTIVITY_TASK_ERROR_COUNTER).inc(1);
+      throw (Error) failure;
+    }
+
+    metricsScope.tagged(activityTypeTag).counter(MetricsType.ACTIVITY_EXEC_FAILED_COUNTER).inc(1);
     RespondActivityTaskFailedRequest result = new RespondActivityTaskFailedRequest();
     failure = CheckedExceptionWrapper.unwrap(failure);
     result.setReason(failure.getClass().getName());

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

@@ -296,6 +296,7 @@ private Promise<byte[]> executeChildWorkflowOnce(
             .setTaskStartToCloseTimeoutSeconds(options.getTaskStartToCloseTimeout().getSeconds())
             .setWorkflowIdReusePolicy(options.getWorkflowIdReusePolicy())
             .setRetryParameters(retryParameters)
+            .setCronSchedule(options.getCronSchedule())
             .build();
     CompletablePromise<byte[]> result = Workflow.newPromise();
     Consumer<Exception> cancellationCallback =

src/main/java/com/uber/cadence/internal/testservice/StateMachines.java

@@ -446,7 +446,8 @@ private static void initiateChildWorkflow(
             .setWorkflowId(d.getWorkflowId())
             .setWorkflowIdReusePolicy(d.getWorkflowIdReusePolicy())
             .setWorkflowType(d.getWorkflowType())
-            .setRetryPolicy(d.getRetryPolicy());
+            .setRetryPolicy(d.getRetryPolicy())
+            .setCronSchedule(d.getCronSchedule());
     HistoryEvent event =
         new HistoryEvent()
             .setEventType(EventType.StartChildWorkflowExecutionInitiated)
@@ -466,7 +467,8 @@ private static void initiateChildWorkflow(
                   .setWorkflowId(d.getWorkflowId())
                   .setWorkflowIdReusePolicy(d.getWorkflowIdReusePolicy())
                   .setWorkflowType(d.getWorkflowType())
-                  .setRetryPolicy(d.getRetryPolicy());
+                  .setRetryPolicy(d.getRetryPolicy())
+                  .setCronSchedule(d.getCronSchedule());
           if (d.isSetInput()) {
             startChild.setInput(d.getInput());
           }

src/main/java/com/uber/cadence/internal/testservice/TestWorkflowMutableStateImpl.java

@@ -344,11 +344,31 @@ private void processDecision(
         processRecordMarker(ctx, d.getRecordMarkerDecisionAttributes(), decisionTaskCompletedId);
         break;
       case RequestCancelExternalWorkflowExecution:
-        throw new InternalServiceError(
-            "Decision " + d.getDecisionType() + " is not yet " + "implemented");
+        processRequestCancelExternalWorkflowExecution(
+            ctx, d.getRequestCancelExternalWorkflowExecutionDecisionAttributes());
+        break;
     }
   }
 
+  private void processRequestCancelExternalWorkflowExecution(
+      RequestContext ctx, RequestCancelExternalWorkflowExecutionDecisionAttributes attr) {
+    ForkJoinPool.commonPool()
+        .execute(
+            () -> {
+              RequestCancelWorkflowExecutionRequest request =
+                  new RequestCancelWorkflowExecutionRequest();
+              WorkflowExecution workflowExecution = new WorkflowExecution();
+              workflowExecution.setWorkflowId(attr.workflowId);
+              request.setWorkflowExecution(workflowExecution);
+              request.setDomain(ctx.getDomain());
+              try {
+                service.RequestCancelWorkflowExecution(request);
+              } catch (Exception e) {
+                log.error("Failure to request cancel external workflow", e);
+              }
+            });
+  }
+
   private void processRecordMarker(
       RequestContext ctx, RecordMarkerDecisionAttributes attr, long decisionTaskCompletedId)
       throws BadRequestError {

src/main/java/com/uber/cadence/internal/testservice/TestWorkflowService.java

@@ -18,7 +18,6 @@
 package com.uber.cadence.internal.testservice;
 
 import com.uber.cadence.BadRequestError;
-import com.uber.cadence.CancellationAlreadyRequestedError;
 import com.uber.cadence.DeprecateDomainRequest;
 import com.uber.cadence.DescribeDomainRequest;
 import com.uber.cadence.DescribeDomainResponse;
@@ -476,8 +475,7 @@ public void RespondActivityTaskCanceledByID(
 
   @Override
   public void RequestCancelWorkflowExecution(RequestCancelWorkflowExecutionRequest cancelRequest)
-      throws BadRequestError, InternalServiceError, EntityNotExistsError,
-          CancellationAlreadyRequestedError, ServiceBusyError, TException {
+      throws TException {
     ExecutionId executionId =
         new ExecutionId(cancelRequest.getDomain(), cancelRequest.getWorkflowExecution());
     TestWorkflowMutableState mutableState = getMutableState(executionId);
@@ -486,8 +484,7 @@ public void RequestCancelWorkflowExecution(RequestCancelWorkflowExecutionRequest
 
   @Override
   public void SignalWorkflowExecution(SignalWorkflowExecutionRequest signalRequest)
-      throws BadRequestError, InternalServiceError, EntityNotExistsError, ServiceBusyError,
-          TException {
+      throws TException {
     ExecutionId executionId =
         new ExecutionId(signalRequest.getDomain(), signalRequest.getWorkflowExecution());
     TestWorkflowMutableState mutableState = getMutableState(executionId);