Versioning Override support (temporalio#2530)

Author: Sushisource

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

@@ -66,6 +66,7 @@ public static WorkflowOptions merge(
         .setLinks(o.getLinks())
         .setOnConflictOptions(o.getOnConflictOptions())
         .setPriority(o.getPriority())
+        .setVersioningOverride(o.getVersioningOverride())
         .validateBuildWithDefaults();
   }
 
@@ -115,6 +116,8 @@ public static final class Builder {
 
     private Priority priority;
 
+    private VersioningOverride versioningOverride;
+
     private Builder() {}
 
     private Builder(WorkflowOptions options) {
@@ -143,6 +146,7 @@ private Builder(WorkflowOptions options) {
       this.links = options.links;
       this.onConflictOptions = options.onConflictOptions;
       this.priority = options.priority;
+      this.versioningOverride = options.versioningOverride;
     }
 
     /**
@@ -472,6 +476,13 @@ public Builder setPriority(Priority priority) {
       return this;
     }
 
+    /** Sets the versioning override to use when starting this workflow. */
+    @Experimental
+    public Builder setVersioningOverride(VersioningOverride versioningOverride) {
+      this.versioningOverride = versioningOverride;
+      return this;
+    }
+
     public WorkflowOptions build() {
       return new WorkflowOptions(
           workflowId,
@@ -495,7 +506,8 @@ public WorkflowOptions build() {
           completionCallbacks,
           links,
           onConflictOptions,
-          priority);
+          priority,
+          versioningOverride);
     }
 
     /**
@@ -524,7 +536,8 @@ public WorkflowOptions validateBuildWithDefaults() {
           completionCallbacks,
           links,
           onConflictOptions,
-          priority);
+          priority,
+          versioningOverride);
     }
   }
 
@@ -569,6 +582,7 @@ public WorkflowOptions validateBuildWithDefaults() {
   private final List<Link> links;
   private final OnConflictOptions onConflictOptions;
   private final Priority priority;
+  private final VersioningOverride versioningOverride;
 
   private WorkflowOptions(
       String workflowId,
@@ -592,7 +606,8 @@ private WorkflowOptions(
       List<Callback> completionCallbacks,
       List<Link> links,
       OnConflictOptions onConflictOptions,
-      Priority priority) {
+      Priority priority,
+      VersioningOverride versioningOverride) {
     this.workflowId = workflowId;
     this.workflowIdReusePolicy = workflowIdReusePolicy;
     this.workflowRunTimeout = workflowRunTimeout;
@@ -615,6 +630,7 @@ private WorkflowOptions(
     this.links = links;
     this.onConflictOptions = onConflictOptions;
     this.priority = priority;
+    this.versioningOverride = versioningOverride;
   }
 
   public String getWorkflowId() {
@@ -721,6 +737,11 @@ public Priority getPriority() {
     return priority;
   }
 
+  @Experimental
+  public VersioningOverride getVersioningOverride() {
+    return versioningOverride;
+  }
+
   public Builder toBuilder() {
     return new Builder(this);
   }
@@ -751,7 +772,8 @@ public boolean equals(Object o) {
         && Objects.equal(completionCallbacks, that.completionCallbacks)
         && Objects.equal(links, that.links)
         && Objects.equal(onConflictOptions, that.onConflictOptions)
-        && Objects.equal(priority, that.priority);
+        && Objects.equal(priority, that.priority)
+        && Objects.equal(versioningOverride, that.versioningOverride);
   }
 
   @Override
@@ -778,7 +800,8 @@ public int hashCode() {
         completionCallbacks,
         links,
         onConflictOptions,
-        priority);
+        priority,
+        versioningOverride);
   }
 
   @Override
@@ -831,6 +854,8 @@ public String toString() {
         + onConflictOptions
         + ", priority="
         + priority
+        + ", versioningOverride="
+        + versioningOverride
         + '}';
   }
 }

temporal-sdk/src/main/java/io/temporal/common/VersioningOverride.java

@@ -0,0 +1,35 @@
+package io.temporal.common;
+
+import javax.annotation.Nonnull;
+
+/** Represents the override of a worker's versioning behavior for a workflow execution. */
+@Experimental
+public abstract class VersioningOverride {
+  private VersioningOverride() {}
+
+  /** Workflow will be pinned to a specific deployment version. */
+  public static final class PinnedVersioningOverride extends VersioningOverride {
+    private final WorkerDeploymentVersion version;
+
+    /**
+     * Creates a new PinnedVersioningOverride.
+     *
+     * @param version The worker deployment version to pin the workflow to.
+     */
+    public PinnedVersioningOverride(@Nonnull WorkerDeploymentVersion version) {
+      this.version = version;
+    }
+
+    /**
+     * @return The worker deployment version to pin the workflow to.
+     */
+    public WorkerDeploymentVersion getVersion() {
+      return version;
+    }
+  }
+
+  /** The workflow will auto-upgrade to the current deployment version on the next workflow task. */
+  public static final class AutoUpgradeVersioningOverride extends VersioningOverride {
+    public AutoUpgradeVersioningOverride() {}
+  }
+}

temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityInfoImpl.java

@@ -5,7 +5,7 @@
 import io.temporal.api.common.v1.Payloads;
 import io.temporal.api.workflowservice.v1.PollActivityTaskQueueResponseOrBuilder;
 import io.temporal.common.Priority;
-import io.temporal.internal.common.PriorityUtils;
+import io.temporal.internal.common.ProtoConverters;
 import io.temporal.internal.common.ProtobufTimeUtils;
 import io.temporal.workflow.Functions;
 import java.time.Duration;
@@ -138,7 +138,7 @@ public boolean isLocal() {
 
   @Override
   public Priority getPriority() {
-    return PriorityUtils.fromProto(response.getPriority());
+    return ProtoConverters.fromProto(response.getPriority());
   }
 
   @Override

temporal-sdk/src/main/java/io/temporal/internal/client/ScheduleProtoUtil.java

@@ -24,7 +24,7 @@
 import io.temporal.common.converter.DataConverter;
 import io.temporal.common.converter.EncodedValues;
 import io.temporal.internal.client.external.GenericWorkflowClient;
-import io.temporal.internal.common.PriorityUtils;
+import io.temporal.internal.common.ProtoConverters;
 import io.temporal.internal.common.ProtobufTimeUtils;
 import io.temporal.internal.common.RetryOptionsUtils;
 import io.temporal.internal.common.SearchAttributesUtil;
@@ -160,7 +160,11 @@ public ScheduleAction actionToProto(io.temporal.client.schedules.ScheduleAction
       workflowRequest.setHeader(grpcHeader);
 
       if (wfOptions.getPriority() != null) {
-        workflowRequest.setPriority(PriorityUtils.toProto(wfOptions.getPriority()));
+        workflowRequest.setPriority(ProtoConverters.toProto(wfOptions.getPriority()));
+      }
+      if (wfOptions.getVersioningOverride() != null) {
+        workflowRequest.setVersioningOverride(
+            ProtoConverters.toProto(wfOptions.getVersioningOverride()));
       }
 
       return ScheduleAction.newBuilder().setStartWorkflow(workflowRequest.build()).build();
@@ -466,7 +470,7 @@ public io.temporal.client.schedules.ScheduleAction protoToAction(@Nonnull Schedu
       }
 
       if (startWfAction.hasPriority()) {
-        wfOptionsBuilder.setPriority(PriorityUtils.fromProto(startWfAction.getPriority()));
+        wfOptionsBuilder.setPriority(ProtoConverters.fromProto(startWfAction.getPriority()));
       }
 
       builder.setOptions(wfOptionsBuilder.build());

temporal-sdk/src/main/java/io/temporal/internal/client/WorkflowClientRequestFactory.java

@@ -19,7 +19,7 @@
 import io.temporal.client.WorkflowOptions;
 import io.temporal.common.RetryOptions;
 import io.temporal.common.context.ContextPropagator;
-import io.temporal.internal.common.PriorityUtils;
+import io.temporal.internal.common.ProtoConverters;
 import io.temporal.internal.common.ProtobufTimeUtils;
 import io.temporal.internal.common.SearchAttributesUtil;
 import java.util.*;
@@ -119,7 +119,11 @@ StartWorkflowExecutionRequest.Builder newStartWorkflowExecutionRequest(
     }
 
     if (options.getPriority() != null) {
-      request.setPriority(PriorityUtils.toProto(options.getPriority()));
+      request.setPriority(ProtoConverters.toProto(options.getPriority()));
+    }
+
+    if (options.getVersioningOverride() != null) {
+      request.setVersioningOverride(ProtoConverters.toProto(options.getVersioningOverride()));
     }
 
     if (options.getSearchAttributes() != null && !options.getSearchAttributes().isEmpty()) {
@@ -202,6 +206,10 @@ SignalWithStartWorkflowExecutionRequest.Builder newSignalWithStartWorkflowExecut
       request.setUserMetadata(startParameters.getUserMetadata());
     }
 
+    if (startParameters.hasVersioningOverride()) {
+      request.setVersioningOverride(startParameters.getVersioningOverride());
+    }
+
     return request;
   }
 

temporal-sdk/src/main/java/io/temporal/internal/common/PriorityUtils.java

@@ -0,0 +1,56 @@
+package io.temporal.internal.common;
+
+import io.temporal.api.common.v1.Priority;
+import io.temporal.api.enums.v1.VersioningBehavior;
+import io.temporal.common.VersioningOverride;
+import io.temporal.common.WorkerDeploymentVersion;
+import javax.annotation.Nonnull;
+
+public class ProtoConverters {
+  public static Priority toProto(io.temporal.common.Priority priority) {
+    return Priority.newBuilder().setPriorityKey(priority.getPriorityKey()).build();
+  }
+
+  @Nonnull
+  public static io.temporal.common.Priority fromProto(@Nonnull Priority priority) {
+    return io.temporal.common.Priority.newBuilder()
+        .setPriorityKey(priority.getPriorityKey())
+        .build();
+  }
+
+  public static io.temporal.api.deployment.v1.WorkerDeploymentVersion toProto(
+      WorkerDeploymentVersion v) {
+    return io.temporal.api.deployment.v1.WorkerDeploymentVersion.newBuilder()
+        .setBuildId(v.getBuildId())
+        .setDeploymentName(v.getDeploymentName())
+        .build();
+  }
+
+  @SuppressWarnings("deprecation")
+  public static io.temporal.api.workflow.v1.VersioningOverride toProto(VersioningOverride v) {
+    if (v instanceof VersioningOverride.PinnedVersioningOverride) {
+      VersioningOverride.PinnedVersioningOverride pv =
+          (VersioningOverride.PinnedVersioningOverride) v;
+      io.temporal.api.workflow.v1.VersioningOverride.PinnedOverride.Builder pinnedBuilder =
+          io.temporal.api.workflow.v1.VersioningOverride.PinnedOverride.newBuilder()
+              .setVersion(toProto(pv.getVersion()));
+
+      pinnedBuilder.setBehavior(
+          io.temporal.api.workflow.v1.VersioningOverride.PinnedOverrideBehavior
+              .PINNED_OVERRIDE_BEHAVIOR_PINNED);
+
+      return io.temporal.api.workflow.v1.VersioningOverride.newBuilder()
+          .setBehavior(VersioningBehavior.VERSIONING_BEHAVIOR_PINNED)
+          .setPinnedVersion(pv.getVersion().toCanonicalString())
+          .setPinned(pinnedBuilder.build())
+          .build();
+    } else {
+      return io.temporal.api.workflow.v1.VersioningOverride.newBuilder()
+          .setBehavior(VersioningBehavior.VERSIONING_BEHAVIOR_AUTO_UPGRADE)
+          .setAutoUpgrade(true)
+          .build();
+    }
+  }
+
+  private ProtoConverters() {}
+}

temporal-sdk/src/main/java/io/temporal/internal/common/ProtoConverters.java

@@ -260,12 +260,12 @@ private Result failureToWFTResult(
       throws Exception {
     String workflowType = workflowTask.getWorkflowType().getName();
     if (e instanceof WorkflowExecutionException) {
+      @SuppressWarnings("deprecation")
       RespondWorkflowTaskCompletedRequest response =
           RespondWorkflowTaskCompletedRequest.newBuilder()
               .setTaskToken(workflowTask.getTaskToken())
               .setIdentity(options.getIdentity())
               .setNamespace(namespace)
-              // TODO: Set stamp or not based on capabilities
               .setBinaryChecksum(options.getBuildId())
               .addCommands(
                   Command.newBuilder()

temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflowTaskHandler.java

@@ -357,6 +357,7 @@ private void handleSingleEventLookahead(HistoryEvent event) {
       case EVENT_TYPE_WORKFLOW_TASK_COMPLETED:
         WorkflowTaskCompletedEventAttributes completedEvent =
             event.getWorkflowTaskCompletedEventAttributes();
+        @SuppressWarnings("deprecation")
         String maybeBuildId = completedEvent.getWorkerVersion().getBuildId();
         if (!maybeBuildId.isEmpty()) {
           currentTaskBuildId = maybeBuildId;

temporal-sdk/src/main/java/io/temporal/internal/statemachines/WorkflowStateMachines.java

@@ -608,7 +608,7 @@ private ExecuteActivityParameters constructExecuteActivityParameters(
         makeUserMetaData(options.getSummary(), null, dataConverterWithCurrentWorkflowContext);
 
     if (options.getPriority() != null) {
-      attributes.setPriority(PriorityUtils.toProto(options.getPriority()));
+      attributes.setPriority(ProtoConverters.toProto(options.getPriority()));
     }
 
     return new ExecuteActivityParameters(attributes, options.getCancellationType(), userMetadata);
@@ -935,7 +935,7 @@ private StartChildWorkflowExecutionParameters createChildWorkflowParameters(
                   replayContext.getTaskQueue().equals(options.getTaskQueue())));
     }
     if (options.getPriority() != null) {
-      attributes.setPriority(PriorityUtils.toProto(options.getPriority()));
+      attributes.setPriority(ProtoConverters.toProto(options.getPriority()));
     }
     return new StartChildWorkflowExecutionParameters(
         attributes, options.getCancellationType(), metadata);