CaN USE_RAMPING_VERSION versioning behaviour (#2868)

Author: THardy98

.github/workflows/ci.yml

@@ -84,7 +84,7 @@ jobs:
 
       - name: Start containerized server and dependencies
         env:
-          TEMPORAL_CLI_VERSION: 1.6.1-server-1.31.0-151.0
+          TEMPORAL_CLI_VERSION: 1.7.0
         run: |
           wget -O temporal_cli.tar.gz https://github.com/temporalio/cli/releases/download/v${TEMPORAL_CLI_VERSION}/temporal_cli_${TEMPORAL_CLI_VERSION}_linux_amd64.tar.gz
           tar -xzf temporal_cli.tar.gz

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

@@ -11,5 +11,26 @@ public enum InitialVersioningBehavior {
    * 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
+  AUTO_UPGRADE,
+
+  /**
+   * Use the Ramping Version of the workflow's task queue at start time, regardless of the
+   * workflow's Target Version.
+   *
+   * <p>After the first workflow task completes, the workflow uses whatever {@link
+   * VersioningBehavior} is specified in the workflow code. If there is no Ramping Version by the
+   * time that the first workflow task is dispatched, it is sent to the Current Version.
+   *
+   * <p>It is highly discouraged to use this if the workflow is annotated with {@link
+   * VersioningBehavior#AUTO_UPGRADE} behavior, because this setting only applies to the first task
+   * of the workflow. If, after the first task, the workflow is AutoUpgrade, it behaves like a
+   * normal AutoUpgrade workflow and goes to the Target Version, which may be the Current Version
+   * instead of the Ramping Version.
+   *
+   * <p>Note that if the workflow being continued has a Pinned override, that override is inherited
+   * by the new workflow run regardless of the {@link InitialVersioningBehavior} specified in the
+   * continue-as-new command. Versioning Override always takes precedence until it is removed
+   * manually via UpdateWorkflowExecutionOptions.
+   */
+  USE_RAMPING_VERSION
 }

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

@@ -1422,6 +1422,11 @@ public void continueAsNew(ContinueAsNewInput input) {
                 io.temporal.api.enums.v1.ContinueAsNewVersioningBehavior
                     .CONTINUE_AS_NEW_VERSIONING_BEHAVIOR_AUTO_UPGRADE);
             break;
+          case USE_RAMPING_VERSION:
+            attributes.setInitialVersioningBehavior(
+                io.temporal.api.enums.v1.ContinueAsNewVersioningBehavior
+                    .CONTINUE_AS_NEW_VERSIONING_BEHAVIOR_USE_RAMPING_VERSION);
+            break;
         }
       }
     }

temporal-sdk/src/test/java/io/temporal/worker/WorkerVersioningTest.java

@@ -525,6 +525,96 @@ public String execute(int attempt) {
     }
   }
 
+  @WorkflowInterface
+  public interface ContinueAsNewWithRampingVersionWorkflow {
+    @WorkflowMethod
+    String execute(int attempt);
+
+    @SignalMethod
+    void continueAsNew();
+  }
+
+  public static class TestWorkerVersioningCanUseRampingVersionV1
+      implements ContinueAsNewWithRampingVersionWorkflow {
+    private boolean continueAsNew;
+
+    @Override
+    @WorkflowVersioningBehavior(VersioningBehavior.PINNED)
+    public String execute(int attempt) {
+      if (attempt > 0) {
+        return "v1.0";
+      }
+      Workflow.await(() -> continueAsNew);
+      ContinueAsNewOptions options =
+          ContinueAsNewOptions.newBuilder()
+              .setInitialVersioningBehavior(InitialVersioningBehavior.USE_RAMPING_VERSION)
+              .build();
+      ContinueAsNewWithRampingVersionWorkflow next =
+          Workflow.newContinueAsNewStub(ContinueAsNewWithRampingVersionWorkflow.class, options);
+      next.execute(attempt + 1);
+      throw new RuntimeException("unreachable");
+    }
+
+    @Override
+    public void continueAsNew() {
+      continueAsNew = true;
+    }
+  }
+
+  public static class TestWorkerVersioningCanUseRampingVersionV2
+      implements ContinueAsNewWithRampingVersionWorkflow {
+    @Override
+    @WorkflowVersioningBehavior(VersioningBehavior.PINNED)
+    public String execute(int attempt) {
+      return "v2.0";
+    }
+
+    @Override
+    public void continueAsNew() {}
+  }
+
+  @Test
+  public void testContinueAsNewWithRampingVersion() {
+    assumeTrue("Test Server doesn't support versioning", SDKTestWorkflowRule.useExternalService);
+
+    WorkerDeploymentVersion v1 =
+        new WorkerDeploymentVersion(testWorkflowRule.getDeploymentName(), "1.0");
+    WorkerDeploymentVersion v2 =
+        new WorkerDeploymentVersion(testWorkflowRule.getDeploymentName(), "2.0");
+
+    Worker w1 = testWorkflowRule.newWorkerWithBuildID("1.0");
+    w1.registerWorkflowImplementationTypes(TestWorkerVersioningCanUseRampingVersionV1.class);
+    w1.start();
+
+    Worker w2 = testWorkflowRule.newWorkerWithBuildID("2.0");
+    w2.registerWorkflowImplementationTypes(TestWorkerVersioningCanUseRampingVersionV2.class);
+    w2.start();
+
+    waitUntilWorkerDeploymentVisible(v1);
+    DescribeWorkerDeploymentResponse d1 = waitUntilWorkerDeploymentVisible(v2);
+    SetWorkerDeploymentCurrentVersionResponse currentResp =
+        setCurrentVersion(v1, d1.getConflictToken());
+    waitForRoutingConfigPropagation(v1);
+
+    ContinueAsNewWithRampingVersionWorkflow wf =
+        testWorkflowRule.newWorkflowStubTimeoutOptions(
+            ContinueAsNewWithRampingVersionWorkflow.class, "can-use-ramping-version");
+    WorkflowExecution we = WorkflowClient.start(wf::execute, 0);
+    waitForWorkflowRunningOnVersion(we.getWorkflowId(), "1.0");
+
+    setRampingVersion(v2, 0, currentResp.getConflictToken());
+    waitForRoutingConfigPropagation(v1, v2);
+
+    wf.continueAsNew();
+
+    String result =
+        testWorkflowRule
+            .getWorkflowClient()
+            .newUntypedWorkflowStub(we.getWorkflowId())
+            .getResult(String.class);
+    Assert.assertEquals("v2.0", result);
+  }
+
   @Test
   public void testContinueAsNewWithVersionUpgrade() {
     assumeTrue("Test Server doesn't support versioning", SDKTestWorkflowRule.useExternalService);
@@ -641,6 +731,12 @@ private SetWorkerDeploymentRampingVersionResponse setRampingVersion(
 
   @SuppressWarnings("deprecation")
   private void waitForRoutingConfigPropagation(WorkerDeploymentVersion v) {
+    waitForRoutingConfigPropagation(v, null);
+  }
+
+  @SuppressWarnings("deprecation")
+  private void waitForRoutingConfigPropagation(
+      WorkerDeploymentVersion currentVersion, WorkerDeploymentVersion rampingVersion) {
     Eventually.assertEventually(
         Duration.ofSeconds(15),
         () -> {
@@ -652,14 +748,22 @@ private void waitForRoutingConfigPropagation(WorkerDeploymentVersion v) {
                   .describeWorkerDeployment(
                       DescribeWorkerDeploymentRequest.newBuilder()
                           .setNamespace(testWorkflowRule.getTestEnvironment().getNamespace())
-                          .setDeploymentName(v.getDeploymentName())
+                          .setDeploymentName(currentVersion.getDeploymentName())
                           .build());
           Assert.assertEquals(
-              v.getBuildId(),
+              currentVersion.getBuildId(),
               resp.getWorkerDeploymentInfo()
                   .getRoutingConfig()
                   .getCurrentDeploymentVersion()
                   .getBuildId());
+          if (rampingVersion != null) {
+            Assert.assertEquals(
+                rampingVersion.getBuildId(),
+                resp.getWorkerDeploymentInfo()
+                    .getRoutingConfig()
+                    .getRampingDeploymentVersion()
+                    .getBuildId());
+          }
           // Check routing config update is not in progress
           int state = resp.getWorkerDeploymentInfo().getRoutingConfigUpdateStateValue();
           Assert.assertNotEquals(