Per Activity Method Options (Part I)

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

@@ -175,6 +175,37 @@ public Builder setCancellationType(ActivityCancellationType cancellationType) {
       return this;
     }
 
+    public Builder mergeActivityOptions(ActivityOptions override) {
+      if (override == null) {
+        return this;
+      }
+      this.taskQueue = (override.taskQueue == null) ? this.taskQueue : override.taskQueue;
+      this.heartbeatTimeout =
+          (override.heartbeatTimeout == null) ? this.heartbeatTimeout : override.heartbeatTimeout;
+      this.retryOptions =
+          (override.retryOptions == null) ? this.retryOptions : override.retryOptions;
+      this.scheduleToCloseTimeout =
+          (override.scheduleToCloseTimeout == null)
+              ? this.scheduleToCloseTimeout
+              : override.scheduleToCloseTimeout;
+      this.startToCloseTimeout =
+          (override.startToCloseTimeout == null)
+              ? this.startToCloseTimeout
+              : override.startToCloseTimeout;
+      this.scheduleToStartTimeout =
+          (override.scheduleToStartTimeout == null)
+              ? this.scheduleToStartTimeout
+              : override.scheduleToStartTimeout;
+      this.cancellationType =
+          (override.cancellationType == null) ? this.cancellationType : override.cancellationType;
+      if (this.contextPropagators == null) {
+        this.contextPropagators = override.contextPropagators;
+      } else if (override.contextPropagators != null) {
+        this.contextPropagators.addAll(override.contextPropagators);
+      }
+      return this;
+    }
+
     /**
      * Properties that are set on this builder take precedence over ones found in the annotation.
      */

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

@@ -51,7 +51,7 @@ public static final class Builder {
     private Duration localRetryThreshold;
     private Duration startToCloseTimeout;
     private RetryOptions retryOptions;
-    private boolean doNotIncludeArgumentsIntoMarker;
+    private Boolean doNotIncludeArgumentsIntoMarker;
 
     /** Copy Builder fields from the options. */
     private Builder(LocalActivityOptions options) {
@@ -94,6 +94,31 @@ public Builder setStartToCloseTimeout(Duration timeout) {
       return this;
     }
 
+    public Builder mergeActivityOptions(LocalActivityOptions override) {
+      if (override == null) {
+        return this;
+      }
+      this.scheduleToCloseTimeout =
+          (override.scheduleToCloseTimeout == null)
+              ? this.scheduleToCloseTimeout
+              : override.scheduleToCloseTimeout;
+      this.localRetryThreshold =
+          (override.localRetryThreshold == null)
+              ? this.localRetryThreshold
+              : override.localRetryThreshold;
+      this.startToCloseTimeout =
+          (override.startToCloseTimeout == null)
+              ? this.startToCloseTimeout
+              : override.startToCloseTimeout;
+      this.retryOptions =
+          (override.retryOptions == null) ? this.retryOptions : override.retryOptions;
+      this.doNotIncludeArgumentsIntoMarker =
+          (override.doNotIncludeArgumentsIntoMarker != null)
+              ? override.doNotIncludeArgumentsIntoMarker
+              : this.doNotIncludeArgumentsIntoMarker;
+      return this;
+    }
+
     /**
      * RetryOptions that define how activity is retried in case of failure. Default is null which is
      * no retries.
@@ -156,14 +181,14 @@ public LocalActivityOptions validateAndBuildWithDefaults() {
   private final Duration localRetryThreshold;
   private final Duration startToCloseTimeout;
   private final RetryOptions retryOptions;
-  private boolean doNotIncludeArgumentsIntoMarker;
+  private Boolean doNotIncludeArgumentsIntoMarker;
 
   private LocalActivityOptions(
       Duration startToCloseTimeout,
       Duration localRetryThreshold,
       Duration scheduleToCloseTimeout,
       RetryOptions retryOptions,
-      boolean doNotIncludeArgumentsIntoMarker) {
+      Boolean doNotIncludeArgumentsIntoMarker) {
     this.localRetryThreshold = localRetryThreshold;
     this.scheduleToCloseTimeout = scheduleToCloseTimeout;
     this.startToCloseTimeout = startToCloseTimeout;
@@ -188,7 +213,7 @@ public RetryOptions getRetryOptions() {
   }
 
   public boolean isDoNotIncludeArgumentsIntoMarker() {
-    return doNotIncludeArgumentsIntoMarker;
+    return (doNotIncludeArgumentsIntoMarker == null) ? false : doNotIncludeArgumentsIntoMarker;
   }
 
   public Builder toBuilder() {
@@ -229,7 +254,7 @@ public String toString() {
         + ", retryOptions="
         + retryOptions
         + ", doNotIncludeArgumentsIntoMarker="
-        + doNotIncludeArgumentsIntoMarker
+        + isDoNotIncludeArgumentsIntoMarker()
         + '}';
   }
 }

temporal-sdk/src/main/java/io/temporal/internal/sync/ActivityInvocationHandler.java

@@ -26,26 +26,33 @@
 import io.temporal.workflow.ActivityStub;
 import java.lang.reflect.InvocationHandler;
 import java.lang.reflect.Method;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.function.Function;
 
 @VisibleForTesting
 public class ActivityInvocationHandler extends ActivityInvocationHandlerBase {
   private final ActivityOptions options;
+  private final Map<String, ActivityOptions> activityMethodOptions;
   private final WorkflowOutboundCallsInterceptor activityExecutor;
 
   @VisibleForTesting
   public static InvocationHandler newInstance(
       Class<?> activityInterface,
       ActivityOptions options,
+      Map<String, ActivityOptions> methodOptions,
       WorkflowOutboundCallsInterceptor activityExecutor) {
-    return new ActivityInvocationHandler(activityInterface, activityExecutor, options);
+    return new ActivityInvocationHandler(
+        activityInterface, activityExecutor, options, methodOptions);
   }
 
   private ActivityInvocationHandler(
       Class<?> activityInterface,
       WorkflowOutboundCallsInterceptor activityExecutor,
-      ActivityOptions options) {
+      ActivityOptions options,
+      Map<String, ActivityOptions> methodOptions) {
     this.options = options;
+    this.activityMethodOptions = (methodOptions == null) ? new HashMap<>() : methodOptions;
     this.activityExecutor = activityExecutor;
     init(activityInterface);
   }
@@ -54,10 +61,12 @@ private ActivityInvocationHandler(
   protected Function<Object[], Object> getActivityFunc(
       Method method, MethodRetry methodRetry, String activityName) {
     Function<Object[], Object> function;
-    ActivityOptions mergedOptions =
-        ActivityOptions.newBuilder(options).mergeMethodRetry(methodRetry).build();
-    ActivityStub stub = ActivityStubImpl.newInstance(mergedOptions, activityExecutor);
-
+    ActivityOptions merged =
+        ActivityOptions.newBuilder(options)
+            .mergeActivityOptions(this.activityMethodOptions.get(activityName))
+            .mergeMethodRetry(methodRetry)
+            .build();
+    ActivityStub stub = ActivityStubImpl.newInstance(merged, activityExecutor);
     function =
         (a) -> stub.execute(activityName, method.getReturnType(), method.getGenericReturnType(), a);
     return function;

temporal-sdk/src/main/java/io/temporal/internal/sync/LocalActivityInvocationHandler.java

@@ -19,42 +19,54 @@
 
 package io.temporal.internal.sync;
 
+import com.google.common.annotations.VisibleForTesting;
 import io.temporal.activity.LocalActivityOptions;
 import io.temporal.common.MethodRetry;
 import io.temporal.common.interceptors.WorkflowOutboundCallsInterceptor;
 import io.temporal.workflow.ActivityStub;
 import java.lang.reflect.InvocationHandler;
 import java.lang.reflect.Method;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.function.Function;
 
-class LocalActivityInvocationHandler extends ActivityInvocationHandlerBase {
+@VisibleForTesting
+public class LocalActivityInvocationHandler extends ActivityInvocationHandlerBase {
   private final LocalActivityOptions options;
+  private final Map<String, LocalActivityOptions> activityMethodOptions;
   private final WorkflowOutboundCallsInterceptor activityExecutor;
 
-  static InvocationHandler newInstance(
+  @VisibleForTesting
+  public static InvocationHandler newInstance(
       Class<?> activityInterface,
       LocalActivityOptions options,
+      Map<String, LocalActivityOptions> methodOptions,
       WorkflowOutboundCallsInterceptor activityExecutor) {
-    return new LocalActivityInvocationHandler(activityInterface, activityExecutor, options);
+    return new LocalActivityInvocationHandler(
+        activityInterface, activityExecutor, options, methodOptions);
   }
 
   private LocalActivityInvocationHandler(
       Class<?> activityInterface,
       WorkflowOutboundCallsInterceptor activityExecutor,
-      LocalActivityOptions options) {
+      LocalActivityOptions options,
+      Map<String, LocalActivityOptions> methodOptions) {
     this.options = options;
+    this.activityMethodOptions = (methodOptions == null) ? new HashMap<>() : methodOptions;
     this.activityExecutor = activityExecutor;
     init(activityInterface);
   }
 
+  @VisibleForTesting
   @Override
-  protected Function<Object[], Object> getActivityFunc(
+  public Function<Object[], Object> getActivityFunc(
       Method method, MethodRetry methodRetry, String activityName) {
     Function<Object[], Object> function;
     LocalActivityOptions mergedOptions =
         LocalActivityOptions.newBuilder(options)
+            .mergeActivityOptions(activityMethodOptions.get(activityName))
             .setMethodRetry(methodRetry)
-            .validateAndBuildWithDefaults();
+            .build();
     ActivityStub stub = LocalActivityStubImpl.newInstance(mergedOptions, activityExecutor);
     function =
         (a) -> stub.execute(activityName, method.getReturnType(), method.getGenericReturnType(), a);

temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowInternal.java

@@ -180,24 +180,41 @@ public static long currentTimeMillis() {
    * Creates client stub to activities that implement given interface.
    *
    * @param activityInterface interface type implemented by activities
+   * @param options options that together with the properties of {@link
+   *     io.temporal.activity.ActivityMethod} specify the activity invocation parameters
+   * @param activityMethodOptions activity method-specific invocation parameters
    */
-  public static <T> T newActivityStub(Class<T> activityInterface, ActivityOptions options) {
+  public static <T> T newActivityStub(
+      Class<T> activityInterface,
+      ActivityOptions options,
+      Map<String, ActivityOptions> activityMethodOptions) {
     InvocationHandler invocationHandler =
         ActivityInvocationHandler.newInstance(
-            activityInterface, options, WorkflowInternal.getWorkflowInterceptor());
+            activityInterface,
+            options,
+            activityMethodOptions,
+            WorkflowInternal.getWorkflowInterceptor());
     return ActivityInvocationHandlerBase.newProxy(activityInterface, invocationHandler);
   }
 
   /**
    * Creates client stub to local activities that implement given interface.
    *
    * @param activityInterface interface type implemented by activities
+   * @param options options that together with the properties of {@link
+   *     io.temporal.activity.ActivityMethod} specify the activity invocation parameters
+   * @param activityMethodOptions activity method-specific invocation parameters
    */
   public static <T> T newLocalActivityStub(
-      Class<T> activityInterface, LocalActivityOptions options) {
+      Class<T> activityInterface,
+      LocalActivityOptions options,
+      Map<String, LocalActivityOptions> activityMethodOptions) {
     InvocationHandler invocationHandler =
         LocalActivityInvocationHandler.newInstance(
-            activityInterface, options, WorkflowInternal.getWorkflowInterceptor());
+            activityInterface,
+            options,
+            activityMethodOptions,
+            WorkflowInternal.getWorkflowInterceptor());
     return ActivityInvocationHandlerBase.newProxy(activityInterface, invocationHandler);
   }
 

temporal-sdk/src/main/java/io/temporal/workflow/Workflow.java

@@ -372,23 +372,38 @@ public final class Workflow {
   public static final int DEFAULT_VERSION = WorkflowInternal.DEFAULT_VERSION;
 
   /**
-   * Creates client stub to activities that implement given interface.
+   * Creates client stub to activities that implement given interface. `
+   *
+   * @param activityInterface interface type implemented by activities
+   */
+  public static <T> T newActivityStub(Class<T> activityInterface) {
+    return WorkflowInternal.newActivityStub(activityInterface, null, null);
+  }
+
+  /**
+   * Creates client stub to activities that implement given interface
    *
    * @param activityInterface interface type implemented by activities.
    * @param options options that together with the properties of {@link
-   *     io.temporal.activity.ActivityMethod} specify the activity invocation parameters.
+   *     io.temporal.activity.ActivityMethod} specify the activity invocation parameters
    */
   public static <T> T newActivityStub(Class<T> activityInterface, ActivityOptions options) {
-    return WorkflowInternal.newActivityStub(activityInterface, options);
+    return WorkflowInternal.newActivityStub(activityInterface, options, null);
   }
 
   /**
-   * Creates client stub to activities that implement given interface. `
+   * Creates client stub to activities that implement given interface.
    *
    * @param activityInterface interface type implemented by activities
+   * @param options options that together with the properties of {@link
+   *     io.temporal.activity.ActivityMethod} specify the activity invocation parameters
+   * @param activityMethodOptions activity method-specific invocation parameters
    */
-  public static <T> T newActivityStub(Class<T> activityInterface) {
-    return WorkflowInternal.newActivityStub(activityInterface, null);
+  public static <T> T newActivityStub(
+      Class<T> activityInterface,
+      ActivityOptions options,
+      Map<String, ActivityOptions> activityMethodOptions) {
+    return WorkflowInternal.newActivityStub(activityInterface, options, activityMethodOptions);
   }
 
   /**
@@ -400,29 +415,44 @@ public static ActivityStub newUntypedActivityStub(ActivityOptions options) {
     return WorkflowInternal.newUntypedActivityStub(options);
   }
 
+  /**
+   * Creates client stub to local activities that implement given interface.
+   *
+   * @param activityInterface interface type implemented by activities
+   */
+  public static <T> T newLocalActivityStub(Class<T> activityInterface) {
+    return WorkflowInternal.newLocalActivityStub(activityInterface, null, null);
+  }
+
   /**
    * Creates client stub to local activities that implement given interface. A local activity is
    * similar to a regular activity, but with some key differences: 1. Local activity is scheduled
    * and run by the workflow worker locally. 2. Local activity does not need Temporal server to
    * schedule activity task and does not rely on activity worker. 3. Local activity is for short
    * living activities (usually finishes within seconds). 4. Local activity cannot heartbeat.
    *
-   * @param activityInterface interface type implemented by activities.
+   * @param activityInterface interface type implemented by activities
    * @param options options that together with the properties of {@link
    *     io.temporal.activity.ActivityMethod} specify the activity invocation parameters.
    */
   public static <T> T newLocalActivityStub(
       Class<T> activityInterface, LocalActivityOptions options) {
-    return WorkflowInternal.newLocalActivityStub(activityInterface, options);
+    return WorkflowInternal.newLocalActivityStub(activityInterface, options, null);
   }
 
   /**
    * Creates client stub to local activities that implement given interface.
    *
    * @param activityInterface interface type implemented by activities
+   * @param options options that together with the properties of {@link
+   *     io.temporal.activity.ActivityMethod} specify the activity invocation parameters
+   * @param activityMethodOptions activity method-specific invocation parameters
    */
-  public static <T> T newLocalActivityStub(Class<T> activityInterface) {
-    return WorkflowInternal.newLocalActivityStub(activityInterface, null);
+  public static <T> T newLocalActivityStub(
+      Class<T> activityInterface,
+      LocalActivityOptions options,
+      Map<String, LocalActivityOptions> activityMethodOptions) {
+    return WorkflowInternal.newLocalActivityStub(activityInterface, options, activityMethodOptions);
   }
 
   /**