Priorities for Workflows/Activities (#2453)

Add priority to various options

Author: Sushisource

.github/workflows/ci.yml

@@ -70,10 +70,36 @@ jobs:
 
     - name: Start containerized server and dependencies
       run: |
-        docker compose \
-          -f ./docker/github/docker-compose.yaml \
-          up -d temporal
-    
+        wget https://github.com/temporalio/cli/releases/download/v1.3.1-priority.0/temporal_cli_1.3.1-priority.0_linux_amd64.tar.gz
+        tar -xzf temporal_cli_1.3.1-priority.0_linux_amd64.tar.gz
+        chmod +x temporal
+        ./temporal server start-dev \
+         --headless \
+         --port 7233 \
+         --http-port 7243 \
+         --namespace UnitTest \
+         --search-attribute CustomKeywordField=Keyword \
+         --search-attribute CustomStringField=Text \
+         --search-attribute CustomTextField=Text \
+         --search-attribute CustomIntField=Int \
+         --search-attribute CustomDatetimeField=Datetime \
+         --search-attribute CustomDoubleField=Double \
+         --search-attribute CustomBoolField=Bool \
+         --dynamic-config-value system.forceSearchAttributesCacheRefreshOnRead=true \
+         --dynamic-config-value system.enableActivityEagerExecution=true \
+         --dynamic-config-value system.enableEagerWorkflowStart=true \
+         --dynamic-config-value system.enableExecuteMultiOperation=true \
+         --dynamic-config-value frontend.enableUpdateWorkflowExecutionAsyncAccepted=true \
+         --dynamic-config-value history.MaxBufferedQueryCount=100000 \
+         --dynamic-config-value frontend.workerVersioningDataAPIs=true \
+         --dynamic-config-value worker.buildIdScavengerEnabled=true \
+         --dynamic-config-value frontend.workerVersioningRuleAPIs=true \
+         --dynamic-config-value worker.removableBuildIdDurationSinceDefault=true \
+         --dynamic-config-value matching.useNewMatcher=true \
+         --dynamic-config-value component.callbacks.allowedAddresses='[{"Pattern":"*","AllowInsecure":true}]' \
+         --dynamic-config-value frontend.workerVersioningWorkflowAPIs=true &
+         sleep 10s
+
     - name: Run unit tests
       env:
         USER: unittest

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

@@ -21,6 +21,8 @@
 package io.temporal.activity;
 
 import io.temporal.api.common.v1.Payloads;
+import io.temporal.common.Experimental;
+import io.temporal.common.Priority;
 import java.time.Duration;
 import java.util.Optional;
 import javax.annotation.Nonnull;
@@ -139,4 +141,14 @@ public interface ActivityInfo {
 
   /** Used to determine if the Activity Execution is a local Activity. */
   boolean isLocal();
+
+  /**
+   * Return the priority of the activity task.
+   *
+   * @apiNote If unset or on an older server version, this method will return {@link
+   *     Priority#getDefaultInstance()}.
+   */
+  @Experimental
+  @Nonnull
+  Priority getPriority();
 }

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

@@ -22,10 +22,7 @@
 
 import com.google.common.base.Objects;
 import io.temporal.client.WorkflowClientOptions;
-import io.temporal.common.Experimental;
-import io.temporal.common.MethodRetry;
-import io.temporal.common.RetryOptions;
-import io.temporal.common.VersioningIntent;
+import io.temporal.common.*;
 import io.temporal.common.context.ContextPropagator;
 import io.temporal.failure.CanceledFailure;
 import java.time.Duration;
@@ -66,6 +63,7 @@ public static final class Builder {
     private boolean disableEagerExecution;
     private VersioningIntent versioningIntent;
     private String summary;
+    private Priority priority;
 
     private Builder() {}
 
@@ -84,6 +82,7 @@ private Builder(ActivityOptions options) {
       this.disableEagerExecution = options.disableEagerExecution;
       this.versioningIntent = options.versioningIntent;
       this.summary = options.summary;
+      this.priority = options.priority;
     }
 
     /**
@@ -195,13 +194,13 @@ public Builder setRetryOptions(RetryOptions retryOptions) {
      * could make sense. <br>
      * This is also why there is no equivalent method on {@link LocalActivityOptions}.
      *
-     * @see <a href="https://github.com/temporalio/sdk-java/issues/490">Rejected feature reqest for
-     *     LocalActivityOption#contextPropagators</a>
      * @param contextPropagators specifies the list of context propagators to use during propagation
      *     from a workflow to the activity with these {@link ActivityOptions}. This list overrides
      *     the list specified on {@link
      *     io.temporal.client.WorkflowClientOptions#getContextPropagators()}, {@code null} means no
      *     overriding
+     * @see <a href="https://github.com/temporalio/sdk-java/issues/490">Rejected feature reqest for
+     *     LocalActivityOption#contextPropagators</a>
      */
     public Builder setContextPropagators(List<ContextPropagator> contextPropagators) {
       this.contextPropagators = contextPropagators;
@@ -258,6 +257,18 @@ public Builder setSummary(String summary) {
       return this;
     }
 
+    /**
+     * Optional priority settings that control relative ordering of task processing when tasks are
+     * backed up in a queue.
+     *
+     * <p>Defaults to inheriting priority from the workflow that scheduled the activity.
+     */
+    @Experimental
+    public Builder setPriority(Priority priority) {
+      this.priority = priority;
+      return this;
+    }
+
     public Builder mergeActivityOptions(ActivityOptions override) {
       if (override == null) {
         return this;
@@ -290,6 +301,7 @@ public Builder mergeActivityOptions(ActivityOptions override) {
         this.versioningIntent = override.versioningIntent;
       }
       this.summary = (override.summary == null) ? this.summary : override.summary;
+      this.priority = (override.priority == null) ? this.priority : override.priority;
       return this;
     }
 
@@ -313,7 +325,8 @@ public ActivityOptions build() {
           cancellationType,
           disableEagerExecution,
           versioningIntent,
-          summary);
+          summary,
+          priority);
     }
 
     public ActivityOptions validateAndBuildWithDefaults() {
@@ -330,7 +343,8 @@ public ActivityOptions validateAndBuildWithDefaults() {
           versioningIntent == null
               ? VersioningIntent.VERSIONING_INTENT_UNSPECIFIED
               : versioningIntent,
-          summary);
+          summary,
+          priority);
     }
   }
 
@@ -345,6 +359,7 @@ public ActivityOptions validateAndBuildWithDefaults() {
   private final boolean disableEagerExecution;
   private final VersioningIntent versioningIntent;
   private final String summary;
+  private final Priority priority;
 
   private ActivityOptions(
       Duration heartbeatTimeout,
@@ -357,7 +372,8 @@ private ActivityOptions(
       ActivityCancellationType cancellationType,
       boolean disableEagerExecution,
       VersioningIntent versioningIntent,
-      String summary) {
+      String summary,
+      Priority priority) {
     this.heartbeatTimeout = heartbeatTimeout;
     this.scheduleToStartTimeout = scheduleToStartTimeout;
     this.scheduleToCloseTimeout = scheduleToCloseTimeout;
@@ -369,6 +385,7 @@ private ActivityOptions(
     this.disableEagerExecution = disableEagerExecution;
     this.versioningIntent = versioningIntent;
     this.summary = summary;
+    this.priority = priority;
   }
 
   /**
@@ -443,6 +460,11 @@ public String getSummary() {
     return summary;
   }
 
+  @Experimental
+  public Priority getPriority() {
+    return priority;
+  }
+
   public Builder toBuilder() {
     return new Builder(this);
   }
@@ -462,7 +484,8 @@ public boolean equals(Object o) {
         && Objects.equal(contextPropagators, that.contextPropagators)
         && disableEagerExecution == that.disableEagerExecution
         && versioningIntent == that.versioningIntent
-        && Objects.equal(summary, that.summary);
+        && Objects.equal(summary, that.summary)
+        && Objects.equal(priority, that.priority);
   }
 
   @Override
@@ -478,7 +501,8 @@ public int hashCode() {
         cancellationType,
         disableEagerExecution,
         versioningIntent,
-        summary);
+        summary,
+        priority);
   }
 
   @Override
@@ -507,6 +531,8 @@ public String toString() {
         + versioningIntent
         + ", summary="
         + summary
+        + ", priority="
+        + priority
         + '}';
   }
 }

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

@@ -85,6 +85,7 @@ public static WorkflowOptions merge(
         .setCompletionCallbacks(o.getCompletionCallbacks())
         .setLinks(o.getLinks())
         .setOnConflictOptions(o.getOnConflictOptions())
+        .setPriority(o.getPriority())
         .validateBuildWithDefaults();
   }
 
@@ -132,6 +133,8 @@ public static final class Builder {
 
     private OnConflictOptions onConflictOptions;
 
+    private Priority priority;
+
     private Builder() {}
 
     private Builder(WorkflowOptions options) {
@@ -159,6 +162,7 @@ private Builder(WorkflowOptions options) {
       this.completionCallbacks = options.completionCallbacks;
       this.links = options.links;
       this.onConflictOptions = options.onConflictOptions;
+      this.priority = options.priority;
     }
 
     /**
@@ -380,8 +384,8 @@ public Builder setContextPropagators(@Nullable List<ContextPropagator> contextPr
      *   <li>has available workflow task executor slots
      * </ul>
      *
-     * and such a {@link WorkflowClient} is used to start a workflow, then the first workflow task
-     * could be dispatched on this local worker with the response to the start call if Server
+     * <p>and such a {@link WorkflowClient} is used to start a workflow, then the first workflow
+     * task could be dispatched on this local worker with the response to the start call if Server
      * supports it. This option can be used to disable this mechanism.
      *
      * <p>Default is true
@@ -478,6 +482,16 @@ public Builder setOnConflictOptions(OnConflictOptions onConflictOptions) {
       return this;
     }
 
+    /**
+     * Optional priority settings that control relative ordering of task processing when tasks are
+     * backed up in a queue.
+     */
+    @Experimental
+    public Builder setPriority(Priority priority) {
+      this.priority = priority;
+      return this;
+    }
+
     public WorkflowOptions build() {
       return new WorkflowOptions(
           workflowId,
@@ -500,7 +514,8 @@ public WorkflowOptions build() {
           requestId,
           completionCallbacks,
           links,
-          onConflictOptions);
+          onConflictOptions,
+          priority);
     }
 
     /**
@@ -528,7 +543,8 @@ public WorkflowOptions validateBuildWithDefaults() {
           requestId,
           completionCallbacks,
           links,
-          onConflictOptions);
+          onConflictOptions,
+          priority);
     }
   }
 
@@ -572,6 +588,7 @@ public WorkflowOptions validateBuildWithDefaults() {
 
   private final List<Link> links;
   private final OnConflictOptions onConflictOptions;
+  private final Priority priority;
 
   private WorkflowOptions(
       String workflowId,
@@ -594,7 +611,8 @@ private WorkflowOptions(
       String requestId,
       List<Callback> completionCallbacks,
       List<Link> links,
-      OnConflictOptions onConflictOptions) {
+      OnConflictOptions onConflictOptions,
+      Priority priority) {
     this.workflowId = workflowId;
     this.workflowIdReusePolicy = workflowIdReusePolicy;
     this.workflowRunTimeout = workflowRunTimeout;
@@ -616,6 +634,7 @@ private WorkflowOptions(
     this.completionCallbacks = completionCallbacks;
     this.links = links;
     this.onConflictOptions = onConflictOptions;
+    this.priority = priority;
   }
 
   public String getWorkflowId() {
@@ -717,6 +736,11 @@ public String getStaticDetails() {
     return onConflictOptions;
   }
 
+  @Experimental
+  public Priority getPriority() {
+    return priority;
+  }
+
   public Builder toBuilder() {
     return new Builder(this);
   }
@@ -746,7 +770,8 @@ public boolean equals(Object o) {
         && Objects.equal(requestId, that.requestId)
         && Objects.equal(completionCallbacks, that.completionCallbacks)
         && Objects.equal(links, that.links)
-        && Objects.equal(onConflictOptions, that.onConflictOptions);
+        && Objects.equal(onConflictOptions, that.onConflictOptions)
+        && Objects.equal(priority, that.priority);
   }
 
   @Override
@@ -772,7 +797,8 @@ public int hashCode() {
         requestId,
         completionCallbacks,
         links,
-        onConflictOptions);
+        onConflictOptions,
+        priority);
   }
 
   @Override
@@ -823,6 +849,8 @@ public String toString() {
         + links
         + ", onConflictOptions="
         + onConflictOptions
+        + ", priority="
+        + priority
         + '}';
   }
 }