Added support for retry policies

sdk/src/main/java/com/microsoft/durabletask/DataConverter.java

@@ -4,12 +4,16 @@
 
 import com.google.protobuf.Timestamp;
 
+import javax.annotation.Nullable;
 import java.time.Instant;
 import java.time.temporal.ChronoUnit;
 
 public interface DataConverter {
-    String serialize(Object value);
-    <T> T deserialize(String data, Class<T> target) throws DataConverterException;
+    @Nullable
+    String serialize(@Nullable Object value);
+
+    @Nullable
+    <T> T deserialize(@Nullable String data, Class<T> target) throws DataConverterException;
 
     default DataConverterException wrapConverterException(String message, Throwable cause) {
         return new DataConverterException(message, cause);

sdk/src/main/java/com/microsoft/durabletask/FailureDetails.java

@@ -5,18 +5,24 @@
 import com.google.protobuf.StringValue;
 import com.microsoft.durabletask.protobuf.OrchestratorService.TaskFailureDetails;
 
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import java.util.Objects;
+
 class FailureDetails {
     private final String errorType;
     private final String errorMessage;
     private final String stackTrace;
 
     public FailureDetails(
             String errorType,
-            String errorMessage,
-            String errorDetails) {
+            @Nullable String errorMessage,
+            @Nullable String errorDetails) {
         this.errorType = errorType;
-        this.errorMessage = errorMessage;
         this.stackTrace = errorDetails;
+
+        // Error message can be null for things like NullPointerException but the gRPC contract doesn't allow null
+        this.errorMessage = Objects.requireNonNullElse(errorMessage, "");
     }
 
     public FailureDetails(Exception exception) {
@@ -29,14 +35,17 @@ public FailureDetails(TaskFailureDetails proto) {
         this.stackTrace = proto.getStackTrace().getValue();
     }
 
+    @Nonnull
     public String getErrorType() {
         return this.errorType;
     }
 
+    @Nonnull
     public String getErrorMessage() {
         return this.errorMessage;
     }
 
+    @Nullable
     public String getStackTrace() {
         return this.stackTrace;
     }
@@ -56,7 +65,7 @@ TaskFailureDetails toProto() {
         return TaskFailureDetails.newBuilder()
                 .setErrorType(this.getErrorType())
                 .setErrorMessage(this.getErrorMessage())
-                .setStackTrace(StringValue.of(this.getStackTrace()))
+                .setStackTrace(StringValue.of(Objects.requireNonNullElse(this.getStackTrace(), "")))
                 .build();
     }
 }

sdk/src/main/java/com/microsoft/durabletask/RetryPolicy.java

@@ -0,0 +1,108 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+package com.microsoft.durabletask;
+
+import javax.annotation.Nullable;
+import java.time.Duration;
+import java.util.Objects;
+
+public class RetryPolicy {
+
+    private final int maxNumberOfAttempts;
+    private final Duration firstRetryInterval;
+    private final double backoffCoefficient;
+    private final Duration maxRetryInterval;
+    private final Duration retryTimeout;
+
+    private RetryPolicy(Builder builder) {
+        this.maxNumberOfAttempts = builder.maxNumberOfAttempts;
+        this.firstRetryInterval = builder.firstRetryInterval;
+        this.backoffCoefficient = builder.backoffCoefficient;
+        this.maxRetryInterval = Objects.requireNonNullElse(builder.maxRetryInterval, Duration.ZERO);
+        this.retryTimeout = Objects.requireNonNullElse(builder.retryTimeout, Duration.ZERO);
+    }
+
+    public static Builder newBuilder(int maxNumberOfAttempts, Duration firstRetryInterval) {
+        return new Builder(maxNumberOfAttempts, firstRetryInterval);
+    }
+
+    public int getMaxNumberOfAttempts() {
+        return this.maxNumberOfAttempts;
+    }
+
+    public Duration getFirstRetryInterval() {
+        return this.firstRetryInterval;
+    }
+
+    public double getBackoffCoefficient() {
+        return this.backoffCoefficient;
+    }
+
+    public Duration getMaxRetryInterval() {
+        return this.maxRetryInterval;
+    }
+
+    public Duration getRetryTimeout() {
+        return this.retryTimeout;
+     }
+
+    public static class Builder {
+        private int maxNumberOfAttempts;
+        private Duration firstRetryInterval;
+        private double backoffCoefficient;
+        private Duration maxRetryInterval;
+        private Duration retryTimeout;
+
+        private Builder(int maxNumberOfAttempts, Duration firstRetryInterval) {
+            this.setMaxNumberOfAttempts(maxNumberOfAttempts);
+            this.setFirstRetryInterval(firstRetryInterval);
+        }
+
+        public RetryPolicy build() {
+            return new RetryPolicy(this);
+        }
+
+        public Builder setMaxNumberOfAttempts(int maxNumberOfAttempts) {
+            if (maxNumberOfAttempts <= 0) {
+                throw new IllegalArgumentException("The value for maxNumberOfAttempts must be greater than zero.");
+            }
+            this.maxNumberOfAttempts = maxNumberOfAttempts;
+            return this;
+        }
+
+        public Builder setFirstRetryInterval(Duration firstRetryInterval) {
+            if (firstRetryInterval == null) {
+                throw new IllegalArgumentException("firstRetryInterval cannot be null.");
+            }
+            if (firstRetryInterval.isZero() || firstRetryInterval.isNegative()) {
+                throw new IllegalArgumentException("The value for firstRetryInterval must be greater than zero.");
+            }
+            this.firstRetryInterval = firstRetryInterval;
+            return this;
+        }
+
+        public Builder setBackoffCoefficient(double backoffCoefficient) {
+            if (backoffCoefficient < 1.0) {
+                throw new IllegalArgumentException("The value for backoffCoefficient must be greater or equal to 1.0.");
+            }
+            this.backoffCoefficient = backoffCoefficient;
+            return this;
+        }
+
+        public Builder setMaxRetryInterval(@Nullable Duration maxRetryInterval) {
+            if (maxRetryInterval != null && maxRetryInterval.compareTo(this.firstRetryInterval) < 0) {
+                throw new IllegalArgumentException("The value for maxRetryInterval must be greater than or equal to the value for firstRetryInterval.");
+            }
+            this.maxRetryInterval = maxRetryInterval;
+            return this;
+        }
+
+        public Builder setRetryTimeout(Duration retryTimeout) {
+            if (retryTimeout != null && retryTimeout.compareTo(this.firstRetryInterval) < 0) {
+                throw new IllegalArgumentException("The value for retryTimeout must be greater than or equal to the value for firstRetryInterval.");
+            }
+            this.retryTimeout = retryTimeout;
+            return this;
+        }
+    }
+}

sdk/src/main/java/com/microsoft/durabletask/TaskOptions.java

@@ -0,0 +1,43 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+package com.microsoft.durabletask;
+
+public class TaskOptions {
+    private final RetryPolicy retryPolicy;
+
+    TaskOptions(Builder builder) {
+        this.retryPolicy = builder.retryPolicy;
+    }
+
+    public static TaskOptions fromRetryPolicy(RetryPolicy policy) {
+        return newBuilder().setRetryPolicy(policy).build();
+    }
+
+    boolean hasRetryPolicy() {
+        return this.retryPolicy != null;
+    }
+
+    public RetryPolicy getRetryPolicy() {
+        return this.retryPolicy;
+    }
+
+    public static Builder newBuilder() {
+        return new Builder();
+    }
+
+    public static class Builder {
+        private RetryPolicy retryPolicy;
+
+        private Builder() {
+        }
+
+        public TaskOptions build() {
+            return new TaskOptions(this);
+        }
+
+        public Builder setRetryPolicy(RetryPolicy retryPolicy) {
+            this.retryPolicy = retryPolicy;
+            return this;
+        }
+    }
+}

sdk/src/main/java/com/microsoft/durabletask/TaskOrchestrationContext.java

@@ -2,6 +2,8 @@
 // Licensed under the MIT License.
 package com.microsoft.durabletask;
 
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
 import java.time.Duration;
 import java.time.Instant;
 import java.util.Arrays;
@@ -25,30 +27,51 @@ default Task<Task<?>> anyOf(Task<?>... tasks) {
     void complete(Object output);
     void fail(FailureDetails failureDetails);
 
-    <V> Task<V> callActivity(String name, Object input, Class<V> returnType);
+    <V> Task<V> callActivity(String name, Object input, TaskOptions options, Class<V> returnType);
 
     default Task<Void> callActivity(String name) {
         return this.callActivity(name, null);
     }
 
     default Task<Void> callActivity(String name, Object input) {
-        return this.callActivity(name, input, Void.class);
+        return this.callActivity(name, input, null, Void.class);
     }
 
-    <V> Task<V> callSubOrchestrator(String name, Object input, String instanceId, Class<V> returnType);
+    default <V> Task<V> callActivity(String name, Object input, Class<V> returnType) {
+        return this.callActivity(name, input, null, returnType);
+    }
+
+    default Task<Void> callActivity(String name, Object input, TaskOptions options) {
+        return this.callActivity(name, input, options, Void.class);
+    }
 
     default Task<Void> callSubOrchestrator(String name){
         return this.callSubOrchestrator(name, null);
     }
 
-    default Task<Void> callSubOrchestrator(String name, Object input){
+    default Task<Void> callSubOrchestrator(String name, Object input) {
         return this.callSubOrchestrator(name, input, null);
     }
 
-    default <V>Task<V> callSubOrchestrator(String name, Object input, Class<V> returnType){
+    default <V>Task<V> callSubOrchestrator(String name, Object input, Class<V> returnType) {
         return this.callSubOrchestrator(name, input, null, returnType);
     }
 
+    default <V> Task<V> callSubOrchestrator(String name, Object input, String instanceId, Class<V> returnType) {
+        return this.callSubOrchestrator(name, input, instanceId, null, returnType);
+    }
+
+    default Task<Void> callSubOrchestrator(String name, Object input, String instanceId, TaskOptions options) {
+        return this.callSubOrchestrator(name, input, instanceId, options, Void.class);
+    }
+
+    <V> Task<V> callSubOrchestrator(
+            String name,
+            @Nullable Object input,
+            @Nullable String instanceId,
+            @Nullable TaskOptions options,
+            Class<V> returnType);
+
     <V> Task<V> waitForExternalEvent(String name, Duration timeout, Class<V> dataType) throws TaskCanceledException;
 
     default Task<Void> waitForExternalEvent(String name, Duration timeout) throws TaskCanceledException {