Add upgrade-on-CAN support

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

@@ -0,0 +1,15 @@
+package io.temporal.common;
+
+/**
+ * Specifies the versioning behavior for the first task of a new workflow run started via
+ * continue-as-new.
+ */
+@Experimental
+public enum InitialVersioningBehavior {
+  /**
+   * Start the new run with {@link VersioningBehavior#AUTO_UPGRADE} behavior for the first task,
+   * upgrading to the latest version. After the first workflow task completes, the workflow uses
+   * whatever versioning behavior is specified in the workflow code.
+   */
+  AUTO_UPGRADE
+}

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

@@ -0,0 +1,15 @@
+package io.temporal.common;
+
+/**
+ * Reason(s) why the server suggests a workflow should continue-as-new. Multiple reasons can be true
+ * at the same time.
+ */
+@Experimental
+public enum SuggestContinueAsNewReason {
+  /** Workflow history size is getting too large. */
+  HISTORY_SIZE_TOO_LARGE,
+  /** Workflow history has too many events. */
+  TOO_MANY_HISTORY_EVENTS,
+  /** Workflow's count of completed plus in-flight updates is too large. */
+  TOO_MANY_UPDATES
+}

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

@@ -8,12 +8,14 @@
 import io.temporal.api.sdk.v1.UserMetadata;
 import io.temporal.api.workflowservice.v1.PollWorkflowTaskQueueResponse;
 import io.temporal.common.RetryOptions;
+import io.temporal.common.SuggestContinueAsNewReason;
 import io.temporal.internal.common.SdkFlag;
 import io.temporal.internal.statemachines.*;
 import io.temporal.workflow.Functions;
 import io.temporal.workflow.Functions.Func;
 import io.temporal.workflow.Functions.Func1;
 import java.time.Duration;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Random;
@@ -357,6 +359,17 @@ Integer getVersion(
    */
   boolean isContinueAsNewSuggested();
 
+  /**
+   * @return the reasons why continue-as-new is suggested, or an empty list if not suggested. This
+   *     value changes during the lifetime of a Workflow Execution.
+   */
+  List<SuggestContinueAsNewReason> getSuggestContinueAsNewReasons();
+
+  /**
+   * @return true if the target worker deployment version has changed for this workflow.
+   */
+  boolean isTargetWorkerDeploymentVersionChanged();
+
   /**
    * @return true if cancellation of the workflow is requested.
    */

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

@@ -10,6 +10,7 @@
 import io.temporal.api.history.v1.WorkflowExecutionStartedEventAttributes;
 import io.temporal.api.sdk.v1.UserMetadata;
 import io.temporal.common.RetryOptions;
+import io.temporal.common.SuggestContinueAsNewReason;
 import io.temporal.failure.CanceledFailure;
 import io.temporal.internal.common.ProtobufTimeUtils;
 import io.temporal.internal.common.SdkFlag;
@@ -416,6 +417,16 @@ public boolean isContinueAsNewSuggested() {
     return workflowStateMachines.isContinueAsNewSuggested();
   }
 
+  @Override
+  public List<SuggestContinueAsNewReason> getSuggestContinueAsNewReasons() {
+    return workflowStateMachines.getSuggestContinueAsNewReasons();
+  }
+
+  @Override
+  public boolean isTargetWorkerDeploymentVersionChanged() {
+    return workflowStateMachines.isTargetWorkerDeploymentVersionChanged();
+  }
+
   /*
    * MUTABLE STATE OPERATIONS
    */

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

@@ -20,6 +20,7 @@
 import io.temporal.api.protocol.v1.Message;
 import io.temporal.api.sdk.v1.UserMetadata;
 import io.temporal.api.workflowservice.v1.GetSystemInfoResponse;
+import io.temporal.common.SuggestContinueAsNewReason;
 import io.temporal.failure.CanceledFailure;
 import io.temporal.internal.common.*;
 import io.temporal.internal.history.LocalActivityMarkerUtils;
@@ -88,6 +89,10 @@ enum HandleEventStatus {
 
   private boolean isContinueAsNewSuggested;
 
+  private List<SuggestContinueAsNewReason> suggestContinueAsNewReasons = Collections.emptyList();
+
+  private boolean isTargetWorkerDeploymentVersionChanged;
+
   /**
    * EventId of the last event seen by these state machines. Events earlier than this one will be
    * discarded.
@@ -276,6 +281,14 @@ public boolean isContinueAsNewSuggested() {
     return isContinueAsNewSuggested;
   }
 
+  public List<SuggestContinueAsNewReason> getSuggestContinueAsNewReasons() {
+    return suggestContinueAsNewReasons;
+  }
+
+  public boolean isTargetWorkerDeploymentVersionChanged() {
+    return isTargetWorkerDeploymentVersionChanged;
+  }
+
   public void setReplaying(boolean replaying) {
     this.replaying = replaying;
   }
@@ -1493,7 +1506,9 @@ public void workflowTaskStarted(
         long currentTimeMillis,
         boolean nonProcessedWorkflowTask,
         long historySize,
-        boolean isContinueAsNewSuggested) {
+        boolean isContinueAsNewSuggested,
+        List<SuggestContinueAsNewReason> suggestContinueAsNewReasons,
+        boolean isTargetWorkerDeploymentVersionChanged) {
       setCurrentTimeMillis(currentTimeMillis);
       for (CancellableCommand cancellableCommand : commands) {
         cancellableCommand.handleWorkflowTaskStarted();
@@ -1509,6 +1524,9 @@ public void workflowTaskStarted(
       WorkflowStateMachines.this.lastWFTStartedEventId = startedEventId;
       WorkflowStateMachines.this.historySize = historySize;
       WorkflowStateMachines.this.isContinueAsNewSuggested = isContinueAsNewSuggested;
+      WorkflowStateMachines.this.suggestContinueAsNewReasons = suggestContinueAsNewReasons;
+      WorkflowStateMachines.this.isTargetWorkerDeploymentVersionChanged =
+          isTargetWorkerDeploymentVersionChanged;
 
       eventLoop();
     }

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

@@ -4,6 +4,10 @@
 import io.temporal.api.enums.v1.EventType;
 import io.temporal.api.enums.v1.WorkflowTaskFailedCause;
 import io.temporal.api.history.v1.WorkflowTaskFailedEventAttributes;
+import io.temporal.common.SuggestContinueAsNewReason;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
 import java.util.Objects;
 
 final class WorkflowTaskStateMachine
@@ -32,7 +36,9 @@ void workflowTaskStarted(
         long currentTimeMillis,
         boolean nonProcessedWorkflowTask,
         long historySize,
-        boolean isContinueAsNewSuggested);
+        boolean isContinueAsNewSuggested,
+        List<SuggestContinueAsNewReason> suggestContinueAsNewReasons,
+        boolean isTargetWorkerDeploymentVersionChanged);
 
     void updateRunId(String currentRunId);
   }
@@ -46,6 +52,8 @@ void workflowTaskStarted(
   private long startedEventId;
   private long historySize;
   private boolean isContinueAsNewSuggested;
+  private List<SuggestContinueAsNewReason> suggestContinueAsNewReasons = Collections.emptyList();
+  private boolean isTargetWorkerDeploymentVersionChanged;
 
   public static WorkflowTaskStateMachine newInstance(
       long workflowTaskStartedEventId, Listener listener) {
@@ -103,6 +111,15 @@ private void handleStarted() {
     historySize = currentEvent.getWorkflowTaskStartedEventAttributes().getHistorySizeBytes();
     isContinueAsNewSuggested =
         currentEvent.getWorkflowTaskStartedEventAttributes().getSuggestContinueAsNew();
+    suggestContinueAsNewReasons =
+        convertSuggestContinueAsNewReasons(
+            currentEvent
+                .getWorkflowTaskStartedEventAttributes()
+                .getSuggestContinueAsNewReasonsList());
+    isTargetWorkerDeploymentVersionChanged =
+        currentEvent
+            .getWorkflowTaskStartedEventAttributes()
+            .getTargetWorkerDeploymentVersionChanged();
 
     // The last started event in the history. So no completed is expected.
     if (currentEvent.getEventId() >= workflowTaskStartedEventId && !hasNextEvent) {
@@ -121,7 +138,33 @@ private void handleCompleted() {
         eventTimeOfTheLastWorkflowStartTask,
         lastTaskInHistory,
         historySize,
-        isContinueAsNewSuggested);
+        isContinueAsNewSuggested,
+        suggestContinueAsNewReasons,
+        isTargetWorkerDeploymentVersionChanged);
+  }
+
+  private static List<SuggestContinueAsNewReason> convertSuggestContinueAsNewReasons(
+      List<io.temporal.api.enums.v1.SuggestContinueAsNewReason> protoReasons) {
+    if (protoReasons.isEmpty()) {
+      return Collections.emptyList();
+    }
+    List<SuggestContinueAsNewReason> reasons = new ArrayList<>(protoReasons.size());
+    for (io.temporal.api.enums.v1.SuggestContinueAsNewReason proto : protoReasons) {
+      switch (proto) {
+        case SUGGEST_CONTINUE_AS_NEW_REASON_HISTORY_SIZE_TOO_LARGE:
+          reasons.add(SuggestContinueAsNewReason.HISTORY_SIZE_TOO_LARGE);
+          break;
+        case SUGGEST_CONTINUE_AS_NEW_REASON_TOO_MANY_HISTORY_EVENTS:
+          reasons.add(SuggestContinueAsNewReason.TOO_MANY_HISTORY_EVENTS);
+          break;
+        case SUGGEST_CONTINUE_AS_NEW_REASON_TOO_MANY_UPDATES:
+          reasons.add(SuggestContinueAsNewReason.TOO_MANY_UPDATES);
+          break;
+        default:
+          break;
+      }
+    }
+    return Collections.unmodifiableList(reasons);
   }
 
   private void handleFailed() {

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

@@ -1415,6 +1415,15 @@ public void continueAsNew(ContinueAsNewInput input) {
                 .determineUseCompatibleFlag(
                     replayContext.getTaskQueue().equals(options.getTaskQueue())));
       }
+      if (options.getInitialVersioningBehavior() != null) {
+        switch (options.getInitialVersioningBehavior()) {
+          case AUTO_UPGRADE:
+            attributes.setInitialVersioningBehavior(
+                io.temporal.api.enums.v1.ContinueAsNewVersioningBehavior
+                    .CONTINUE_AS_NEW_VERSIONING_BEHAVIOR_AUTO_UPGRADE);
+            break;
+        }
+      }
     }
 
     if (options == null && replayContext.getRetryOptions() != null) {

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

@@ -4,10 +4,12 @@
 import io.temporal.api.common.v1.WorkflowExecution;
 import io.temporal.common.Priority;
 import io.temporal.common.RetryOptions;
+import io.temporal.common.SuggestContinueAsNewReason;
 import io.temporal.internal.common.ProtoConverters;
 import io.temporal.internal.replay.ReplayWorkflowContext;
 import io.temporal.workflow.WorkflowInfo;
 import java.time.Duration;
+import java.util.List;
 import java.util.Optional;
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
@@ -147,6 +149,16 @@ public boolean isContinueAsNewSuggested() {
     return context.isContinueAsNewSuggested();
   }
 
+  @Override
+  public List<SuggestContinueAsNewReason> getSuggestContinueAsNewReasons() {
+    return context.getSuggestContinueAsNewReasons();
+  }
+
+  @Override
+  public boolean isTargetWorkerDeploymentVersionChanged() {
+    return context.isTargetWorkerDeploymentVersionChanged();
+  }
+
   @Override
   public Optional<String> getCurrentBuildId() {
     return context.getCurrentBuildId();

temporal-sdk/src/main/java/io/temporal/internal/worker/NexusWorker.java

@@ -282,6 +282,7 @@ public Throwable wrapFailure(NexusTask task, Throwable failure) {
           "Failure processing nexus response: " + response.getRequest().toString(), failure);
     }
 
+    @SuppressWarnings("deprecation") // Uses hasOperationError()/getOperationError() for compat
     private void handleNexusTask(NexusTask task, Scope metricsScope) {
       PollNexusTaskQueueResponseOrBuilder pollResponse = task.getResponse();
       ByteString taskToken = pollResponse.getTaskToken();
@@ -374,6 +375,8 @@ private void logExceptionDuringResultReporting(
       }
     }
 
+    @SuppressWarnings(
+        "deprecation") // Uses setOperationError() for backward compat with old servers
     private Response getResponseForOldServer(Response response) {
       Response.Builder b = response.toBuilder();
       Failure failure = response.getStartOperation().getFailure();

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

@@ -45,6 +45,8 @@ public static final class Builder {
     @SuppressWarnings("deprecation")
     private VersioningIntent versioningIntent;
 
+    private InitialVersioningBehavior initialVersioningBehavior;
+
     private Builder() {}
 
     private Builder(ContinueAsNewOptions options) {
@@ -60,6 +62,7 @@ private Builder(ContinueAsNewOptions options) {
       this.typedSearchAttributes = options.getTypedSearchAttributes();
       this.contextPropagators = options.getContextPropagators();
       this.versioningIntent = options.versioningIntent;
+      this.initialVersioningBehavior = options.initialVersioningBehavior;
     }
 
     public Builder setWorkflowRunTimeout(Duration workflowRunTimeout) {
@@ -131,6 +134,18 @@ public Builder setVersioningIntent(VersioningIntent versioningIntent) {
       return this;
     }
 
+    /**
+     * Specifies the versioning behavior for the first task of the new workflow run. For example,
+     * set to AUTO_UPGRADE to upgrade to the latest version on continue-as-new instead of inheriting
+     * the pinned version from the previous run.
+     */
+    @Experimental
+    public Builder setInitialVersioningBehavior(
+        InitialVersioningBehavior initialVersioningBehavior) {
+      this.initialVersioningBehavior = initialVersioningBehavior;
+      return this;
+    }
+
     public ContinueAsNewOptions build() {
       return new ContinueAsNewOptions(
           workflowRunTimeout,
@@ -141,7 +156,8 @@ public ContinueAsNewOptions build() {
           searchAttributes,
           typedSearchAttributes,
           contextPropagators,
-          versioningIntent);
+          versioningIntent,
+          initialVersioningBehavior);
     }
   }
 
@@ -157,6 +173,8 @@ public ContinueAsNewOptions build() {
   @SuppressWarnings("deprecation")
   private final @Nullable VersioningIntent versioningIntent;
 
+  private final @Nullable InitialVersioningBehavior initialVersioningBehavior;
+
   public ContinueAsNewOptions(
       @Nullable Duration workflowRunTimeout,
       @Nullable String taskQueue,
@@ -166,7 +184,8 @@ public ContinueAsNewOptions(
       @Nullable Map<String, Object> searchAttributes,
       @Nullable SearchAttributes typedSearchAttributes,
       @Nullable List<ContextPropagator> contextPropagators,
-      @SuppressWarnings("deprecation") @Nullable VersioningIntent versioningIntent) {
+      @SuppressWarnings("deprecation") @Nullable VersioningIntent versioningIntent,
+      @Nullable InitialVersioningBehavior initialVersioningBehavior) {
     this.workflowRunTimeout = workflowRunTimeout;
     this.taskQueue = taskQueue;
     this.retryOptions = retryOptions;
@@ -176,6 +195,7 @@ public ContinueAsNewOptions(
     this.typedSearchAttributes = typedSearchAttributes;
     this.contextPropagators = contextPropagators;
     this.versioningIntent = versioningIntent;
+    this.initialVersioningBehavior = initialVersioningBehavior;
   }
 
   public @Nullable Duration getWorkflowRunTimeout() {
@@ -223,4 +243,13 @@ public RetryOptions getRetryOptions() {
   public @Nullable VersioningIntent getVersioningIntent() {
     return versioningIntent;
   }
+
+  /**
+   * @return the initial versioning behavior for the first task of the new workflow run, or null if
+   *     unset.
+   */
+  @Experimental
+  public @Nullable InitialVersioningBehavior getInitialVersioningBehavior() {
+    return initialVersioningBehavior;
+  }
 }