StateMachine rewrite

build.gradle

@@ -16,7 +16,7 @@ plugins {
     id 'maven-publish'
     id 'signing'
     id 'de.marcphilipp.nexus-publish' version '0.4.0'
-    id 'name.remal.check-updates' version '1.0.203'
+    id 'name.remal.check-updates' version '1.0.209'
 }
 
 repositories {
@@ -98,15 +98,15 @@ dependencies {
     api group: 'com.uber.m3', name: 'tally-core', version: '0.6.1'
     api group: 'io.micrometer', name: 'micrometer-core', version: '1.5.2'
     api group: 'org.slf4j', name: 'slf4j-api', version: '1.7.30'
-    api 'io.grpc:grpc-protobuf:1.30.2'
-    api 'io.grpc:grpc-stub:1.30.2'
+    api 'io.grpc:grpc-protobuf:1.31.1'
+    api 'io.grpc:grpc-stub:1.31.1'
 
     implementation group: 'com.google.guava', name: 'guava', version: '29.0-jre'
-    implementation group: 'com.cronutils', name: 'cron-utils', version: '9.1.0'
-    implementation 'io.grpc:grpc-netty-shaded:1.30.2'
-    implementation group: 'com.google.protobuf', name: 'protobuf-java-util', version: '3.12.2'
-    implementation group: 'com.fasterxml.jackson.core', name: 'jackson-databind', version: '2.11.1'
-    implementation group: 'com.fasterxml.jackson.datatype', name: 'jackson-datatype-jsr310', version: '2.11.1'
+    implementation group: 'com.cronutils', name: 'cron-utils', version: '9.1.1'
+    implementation 'io.grpc:grpc-netty-shaded:1.31.1'
+    implementation group: 'com.google.protobuf', name: 'protobuf-java-util', version: '3.12.4'
+    implementation group: 'com.fasterxml.jackson.core', name: 'jackson-databind', version: '2.11.2'
+    implementation group: 'com.fasterxml.jackson.datatype', name: 'jackson-datatype-jsr310', version: '2.11.2'
 
     testImplementation group: 'ch.qos.logback', name: 'logback-classic', version: '1.2.3'
     testImplementation group: 'com.googlecode.junit-toolbox', name: 'junit-toolbox', version: '2.4'
@@ -120,7 +120,7 @@ configurations.all {
 
 license {
     header rootProject.file('license-header.txt')
-    exclude 'io/temporal/proto/**.java' // generated code
+    exclude '**/*.puml'
 }
 
 task initSubmodules(type: Exec) {
@@ -143,11 +143,11 @@ jar {
 
 protobuf {
     protoc {
-        artifact = 'com.google.protobuf:protoc:3.12.3'
+        artifact = 'com.google.protobuf:protoc:3.12.4'
     }
     plugins {
         grpc {
-            artifact = 'io.grpc:protoc-gen-grpc-java:1.30.2'
+            artifact = 'io.grpc:protoc-gen-grpc-java:1.31.1'
         }
     }
     generateProtoTasks {
@@ -170,8 +170,8 @@ protobuf {
 
 idea {
     module {
-        sourceDirs += file("$buildDir/generated/main/java");
-        sourceDirs += file("$buildDir/generated/main/grpc");
+        sourceDirs += file("$buildDir/generated/main/java")
+        sourceDirs += file("$buildDir/generated/main/grpc")
     }
 }
 
@@ -327,15 +327,15 @@ allprojects {
     // add a testlistener to all tasks of type Test
     tasks.withType(Test) {
         afterTest { TestDescriptor descriptor, TestResult result ->
-            if(result.resultType == org.gradle.api.tasks.testing.TestResult.ResultType.FAILURE){
+            if (result.resultType == org.gradle.api.tasks.testing.TestResult.ResultType.FAILURE) {
                 failedTests << ["${descriptor.className}::${descriptor.name}"]
             }
         }
     }
 
     // print out tracked failed tests when the build has finished
     gradle.buildFinished {
-        if(!failedTests.empty){
+        if (!failedTests.empty) {
             println "Failed tests for ${project.name}:"
             failedTests.each { failedTest ->
                 println failedTest

gradle/wrapper/gradle-wrapper.properties

@@ -1,5 +1,5 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-6.5.1-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-6.6-bin.zip
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists

gradlew

@@ -82,6 +82,7 @@ esac
 
 CLASSPATH=$APP_HOME/gradle/wrapper/gradle-wrapper.jar
 
+
 # Determine the Java command to use to start the JVM.
 if [ -n "$JAVA_HOME" ] ; then
     if [ -x "$JAVA_HOME/jre/sh/java" ] ; then
@@ -129,6 +130,7 @@ fi
 if [ "$cygwin" = "true" -o "$msys" = "true" ] ; then
     APP_HOME=`cygpath --path --mixed "$APP_HOME"`
     CLASSPATH=`cygpath --path --mixed "$CLASSPATH"`
+
     JAVACMD=`cygpath --unix "$JAVACMD"`
 
     # We build the pattern for arguments to be converted via cygpath

gradlew.bat

@@ -84,6 +84,7 @@ set CMD_LINE_ARGS=%*
 
 set CLASSPATH=%APP_HOME%\gradle\wrapper\gradle-wrapper.jar
 
+
 @rem Execute Gradle
 "%JAVA_EXE%" %DEFAULT_JVM_OPTS% %JAVA_OPTS% %GRADLE_OPTS% "-Dorg.gradle.appname=%APP_BASE_NAME%" -classpath "%CLASSPATH%" org.gradle.wrapper.GradleWrapperMain %CMD_LINE_ARGS%
 

src/main/java/io/temporal/activity/ActivityCancellationType.java

@@ -24,7 +24,7 @@
 
 /**
  * Defines behaviour of the parent workflow when {@link CancellationScope} that wraps child workflow
- * execution request is cancelled. The result of the cancellation independently of the type is a
+ * execution request is canceled. The result of the cancellation independently of the type is a
  * {@link CanceledFailure} thrown from the child workflow method.
  */
 public enum ActivityCancellationType {

src/main/java/io/temporal/activity/ActivityOptions.java

@@ -185,7 +185,7 @@ public ActivityOptions validateAndBuildWithDefaults() {
           taskQueue,
           retryOptions,
           contextPropagators,
-          cancellationType);
+          cancellationType == null ? ActivityCancellationType.TRY_CANCEL : cancellationType);
     }
   }
 

src/main/java/io/temporal/activity/LocalActivityOptions.java

@@ -48,6 +48,7 @@ public static LocalActivityOptions getDefaultInstance() {
 
   public static final class Builder {
     private Duration scheduleToCloseTimeout;
+    private Duration localRetryThreshold;
     private Duration startToCloseTimeout;
     private RetryOptions retryOptions;
 
@@ -57,6 +58,7 @@ private Builder(LocalActivityOptions options) {
         return;
       }
       this.scheduleToCloseTimeout = options.getScheduleToCloseTimeout();
+      this.localRetryThreshold = options.getLocalRetryThreshold();
       this.startToCloseTimeout = options.getStartToCloseTimeout();
       this.retryOptions = options.retryOptions;
     }
@@ -67,6 +69,15 @@ public Builder setScheduleToCloseTimeout(Duration scheduleToCloseTimeout) {
       return this;
     }
 
+    /**
+     * Maximum time to retry locally keeping workflow task open through heartbeat. Default is 6
+     * workflow task timeout.
+     */
+    public Builder setLocalRetryThreshold(Duration localRetryThreshold) {
+      this.localRetryThreshold = localRetryThreshold;
+      return this;
+    }
+
     public Builder setStartToCloseTimeout(Duration startToCloseTimeout) {
       this.startToCloseTimeout = startToCloseTimeout;
       return this;
@@ -93,24 +104,31 @@ public Builder setMethodRetry(MethodRetry r) {
     }
 
     public LocalActivityOptions build() {
-      return new LocalActivityOptions(scheduleToCloseTimeout, startToCloseTimeout, retryOptions);
+      return new LocalActivityOptions(
+          startToCloseTimeout, localRetryThreshold, scheduleToCloseTimeout, retryOptions);
     }
 
     public LocalActivityOptions validateAndBuildWithDefaults() {
       RetryOptions ro = null;
       if (retryOptions != null) {
         ro = RetryOptions.newBuilder(retryOptions).validateBuildWithDefaults();
       }
-      return new LocalActivityOptions(scheduleToCloseTimeout, startToCloseTimeout, ro);
+      return new LocalActivityOptions(
+          startToCloseTimeout, localRetryThreshold, scheduleToCloseTimeout, ro);
     }
   }
 
   private final Duration scheduleToCloseTimeout;
+  private final Duration localRetryThreshold;
   private final Duration startToCloseTimeout;
   private final RetryOptions retryOptions;
 
   private LocalActivityOptions(
-      Duration scheduleToCloseTimeout, Duration startToCloseTimeout, RetryOptions retryOptions) {
+      Duration startToCloseTimeout,
+      Duration localRetryThreshold,
+      Duration scheduleToCloseTimeout,
+      RetryOptions retryOptions) {
+    this.localRetryThreshold = localRetryThreshold;
     this.scheduleToCloseTimeout = scheduleToCloseTimeout;
     this.startToCloseTimeout = startToCloseTimeout;
     this.retryOptions = retryOptions;
@@ -120,6 +138,10 @@ public Duration getScheduleToCloseTimeout() {
     return scheduleToCloseTimeout;
   }
 
+  public Duration getLocalRetryThreshold() {
+    return localRetryThreshold;
+  }
+
   public Duration getStartToCloseTimeout() {
     return startToCloseTimeout;
   }

src/main/java/io/temporal/client/ActivityCancelledException.java → src/main/java/io/temporal/client/ActivityCanceledException.java

@@ -25,13 +25,13 @@
  * Usually indicates that activity was already completed (duplicated request to complete) or timed
  * out or workflow is closed.
  */
-public final class ActivityCancelledException extends ActivityCompletionException {
+public final class ActivityCanceledException extends ActivityCompletionException {
 
-  public ActivityCancelledException(ActivityInfo info) {
+  public ActivityCanceledException(ActivityInfo info) {
     super(info);
   }
 
-  public ActivityCancelledException() {
+  public ActivityCanceledException() {
     super();
   }
 }

src/main/java/io/temporal/client/WorkflowOptions.java

@@ -132,7 +132,7 @@ public Builder setWorkflowId(String workflowId) {
      *
      *     <ul>
      *       AllowDuplicateFailedOnly is a default value. It means that workflow can start if
-     *       previous run failed or was cancelled or terminated.
+     *       previous run failed or was canceled or terminated.
      * </ul>
      *
      * <ul>
@@ -263,15 +263,15 @@ public WorkflowOptions validateBuildWithDefaults() {
 
   private final String taskQueue;
 
-  private RetryOptions retryOptions;
+  private final RetryOptions retryOptions;
 
-  private String cronSchedule;
+  private final String cronSchedule;
 
-  private Map<String, Object> memo;
+  private final Map<String, Object> memo;
 
-  private Map<String, Object> searchAttributes;
+  private final Map<String, Object> searchAttributes;
 
-  private List<ContextPropagator> contextPropagators;
+  private final List<ContextPropagator> contextPropagators;
 
   private WorkflowOptions(
       String workflowId,

src/main/java/io/temporal/client/WorkflowStub.java

@@ -143,7 +143,7 @@ <R> CompletableFuture<R> getResultAsync(
    * Request cancellation of a workflow execution.
    *
    * <p>Cancellation cancels {@link io.temporal.workflow.CancellationScope} that wraps the main
-   * workflow method. Note that workflow can take long time to get cancelled or even completely
+   * workflow method. Note that workflow can take long time to get canceled or even completely
    * ignore the cancellation request.
    */
   void cancel();

src/main/java/io/temporal/common/CronSchedule.java

@@ -31,7 +31,7 @@
  * 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.
+ * stop until it is terminated or canceled.
  */
 @Retention(RetentionPolicy.RUNTIME)
 @Target(ElementType.METHOD)

src/main/java/io/temporal/common/converter/EncodedValues.java

@@ -27,7 +27,7 @@
 public final class EncodedValues implements Values {
   private Optional<Payloads> payloads;
   private DataConverter converter;
-  private Object[] values;
+  private final Object[] values;
 
   public EncodedValues(Optional<Payloads> payloads, DataConverter converter) {
     this.payloads = Objects.requireNonNull(payloads);

src/main/java/io/temporal/common/interceptors/WorkflowOutboundCallsInterceptor.java

@@ -129,4 +129,6 @@ void registerSignal(
   void upsertSearchAttributes(Map<String, Object> searchAttributes);
 
   Object newThread(Runnable runnable, boolean detached, String name);
+
+  long currentTimeMillis();
 }

src/main/java/io/temporal/common/interceptors/WorkflowOutboundCallsInterceptorBase.java

@@ -166,4 +166,9 @@ public void upsertSearchAttributes(Map<String, Object> searchAttributes) {
   public Object newThread(Runnable runnable, boolean detached, String name) {
     return next.newThread(runnable, detached, name);
   }
+
+  @Override
+  public long currentTimeMillis() {
+    return next.currentTimeMillis();
+  }
 }

src/main/java/io/temporal/failure/ApplicationFailure.java

@@ -120,7 +120,7 @@ public void setDataConverter(DataConverter converter) {
   }
 
   private static String getMessage(String message, String type, boolean nonRetryable) {
-    return (Strings.isNullOrEmpty(message) ? "" : "message='" + message + "\', ")
+    return (Strings.isNullOrEmpty(message) ? "" : "message='" + message + "', ")
         + "type='"
         + type
         + '\''

src/main/java/io/temporal/failure/CanceledFailure.java

@@ -22,13 +22,14 @@
 import io.temporal.common.converter.DataConverter;
 import io.temporal.common.converter.EncodedValues;
 import io.temporal.common.converter.Values;
+import java.util.Objects;
 
 public final class CanceledFailure extends TemporalFailure {
   private final Values details;
 
   public CanceledFailure(String message, Values details, Throwable cause) {
     super(message, message, cause);
-    this.details = details;
+    this.details = Objects.requireNonNull(details);
   }
 
   public CanceledFailure(String message, Object details) {

src/main/java/io/temporal/failure/FailureConverter.java

@@ -33,7 +33,7 @@
 import io.temporal.api.failure.v1.ServerFailureInfo;
 import io.temporal.api.failure.v1.TerminatedFailureInfo;
 import io.temporal.api.failure.v1.TimeoutFailureInfo;
-import io.temporal.client.ActivityCancelledException;
+import io.temporal.client.ActivityCanceledException;
 import io.temporal.common.converter.DataConverter;
 import io.temporal.common.converter.EncodedValues;
 import io.temporal.internal.common.CheckedExceptionWrapper;
@@ -66,11 +66,11 @@ public class FailureConverter {
 
   private static final Pattern TRACE_ELEMENT_PATTERN = Pattern.compile(TRACE_ELEMENT_REGEXP);
 
-  public static Exception failureToException(Failure failure, DataConverter dataConverter) {
+  public static RuntimeException failureToException(Failure failure, DataConverter dataConverter) {
     if (failure == null) {
       return null;
     }
-    Exception result = failureToExceptionImpl(failure, dataConverter);
+    RuntimeException result = failureToExceptionImpl(failure, dataConverter);
     if (result instanceof TemporalFailure) {
       ((TemporalFailure) result).setFailure(failure);
     }
@@ -81,8 +81,9 @@ public static Exception failureToException(Failure failure, DataConverter dataCo
     return result;
   }
 
-  private static Exception failureToExceptionImpl(Failure failure, DataConverter dataConverter) {
-    Exception cause =
+  private static RuntimeException failureToExceptionImpl(
+      Failure failure, DataConverter dataConverter) {
+    RuntimeException cause =
         failure.hasCause() ? failureToException(failure.getCause(), dataConverter) : null;
     switch (failure.getFailureInfoCase()) {
       case APPLICATION_FAILURE_INFO:
@@ -257,7 +258,7 @@ public static Failure exceptionToFailureNoUnwrapping(Throwable e) {
               .setWorkflowType(WorkflowType.newBuilder().setName(ce.getWorkflowType()))
               .setWorkflowExecution(ce.getExecution());
       failure.setChildWorkflowExecutionFailureInfo(info);
-    } else if (e instanceof ActivityCancelledException) {
+    } else if (e instanceof ActivityCanceledException) {
       CanceledFailureInfo.Builder info = CanceledFailureInfo.newBuilder();
       failure.setCanceledFailureInfo(info);
     } else {