Merge branch 'master' into worker-heartbeat-final

Author: yuandrew

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;
@@ -52,7 +53,7 @@ enum HandleEventStatus {
   /** Initial set of SDK flags that will be set on all new workflow executions. */
   @VisibleForTesting
   public static List<SdkFlag> initialFlags =
-      Collections.unmodifiableList(Arrays.asList(SdkFlag.SKIP_YIELD_ON_DEFAULT_VERSION));
+      Collections.singletonList(SdkFlag.SKIP_YIELD_ON_DEFAULT_VERSION);
 
   /**
    * Keep track of the change versions that have been seen by the SDK. This is used to generate the
@@ -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

@@ -1174,7 +1174,7 @@ public int getVersion(String changeId, int minSupported, int maxSupported) {
      * Previously the SDK would yield on the getVersion call to the scheduler. This is not ideal because it can lead to non-deterministic
      * scheduling if the getVersion call was removed.
      * */
-    if (replayContext.checkSdkFlag(SdkFlag.SKIP_YIELD_ON_VERSION)) {
+    if (replayContext.tryUseSdkFlag(SdkFlag.SKIP_YIELD_ON_VERSION)) {
       // This can happen if we are replaying a workflow and encounter a getVersion call that did not
       // exist on the original execution and the range does not include the default version.
       if (versionToUse == null) {
@@ -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/ExecutorThreadFactory.java

@@ -7,18 +7,23 @@ class ExecutorThreadFactory implements ThreadFactory {
   private final String threadPrefix;
 
   private final Thread.UncaughtExceptionHandler uncaughtExceptionHandler;
+  private final ClassLoader contextClassLoader;
   private final AtomicInteger threadIndex = new AtomicInteger();
 
   public ExecutorThreadFactory(String threadPrefix, Thread.UncaughtExceptionHandler eh) {
     this.threadPrefix = threadPrefix;
     this.uncaughtExceptionHandler = eh;
+    this.contextClassLoader = Thread.currentThread().getContextClassLoader();
   }
 
   @Override
   public Thread newThread(Runnable r) {
     Thread result = new Thread(r);
     result.setName(threadPrefix + ": " + threadIndex.incrementAndGet());
     result.setUncaughtExceptionHandler(uncaughtExceptionHandler);
+    if (contextClassLoader != null) {
+      result.setContextClassLoader(contextClassLoader);
+    }
     return result;
   }
 }

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

@@ -315,10 +315,7 @@ public Throwable wrapFailure(NexusTask task, Throwable failure) {
           "Failure processing nexus response: " + response.getRequest().toString(), failure);
     }
 
-    /**
-     * @return true if the handler reported a failure or error
-     */
-    @SuppressWarnings("deprecation") // Uses deprecated operationError
+    @SuppressWarnings("deprecation") // Uses hasOperationError()/getOperationError() for compat
     private boolean handleNexusTask(NexusTask task, Scope metricsScope) {
       PollNexusTaskQueueResponseOrBuilder pollResponse = task.getResponse();
       ByteString taskToken = pollResponse.getTaskToken();