Add versioning to sub orchestrations

Signed-off-by: Hal Spang <halspang@microsoft.com>

Author: halspang

client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcWorker.java

@@ -117,7 +117,8 @@ public void startAndBlock() {
                 this.orchestrationFactories,
                 this.dataConverter,
                 this.maximumTimerInterval,
-                logger);
+                logger,
+                this.versioningOptions);
         TaskActivityExecutor taskActivityExecutor = new TaskActivityExecutor(
                 this.activityFactories,
                 this.dataConverter,

client/src/main/java/com/microsoft/durabletask/NewSubOrchestrationInstanceOptions.java

@@ -0,0 +1,62 @@
+package com.microsoft.durabletask;
+
+/**
+ * Options for starting a new sub-orchestration instance.
+ */
+public class NewSubOrchestrationInstanceOptions extends TaskOptions {
+
+    private String instanceId;
+    private String version;
+
+    /**
+     * Creates options with a retry policy for the sub-orchestration.
+     * @param retryPolicy The retry policy to use for the sub-orchestration.
+     */
+    public NewSubOrchestrationInstanceOptions(RetryPolicy retryPolicy) {
+        super(retryPolicy);
+    }
+
+    /**
+     * Creates options with a retry handler for the sub-orchestration.
+     * @param retryHandler The retry handler to use for the sub-orchestration.
+     */
+    public NewSubOrchestrationInstanceOptions(RetryHandler retryHandler) {
+        super(retryHandler);
+    }
+
+    /**
+     * Sets the version for the sub-orchestration instance.
+     * @param version The version string to use.
+     * @return This options object for chaining.
+     */
+    public NewSubOrchestrationInstanceOptions setVersion(String version) {
+        this.version = version;
+        return this;
+    }
+
+    /**
+     * Gets the version for the sub-orchestration instance.
+     * @return The version string, or null if not set.
+     */
+    public String getVersion() {
+        return version;
+    }
+
+    /**
+     * Sets a custom instance ID for the sub-orchestration.
+     * @param instanceId The custom instance ID to use.
+     * @return This options object for chaining.
+     */
+    public NewSubOrchestrationInstanceOptions setInstanceId(String instanceId) {
+        this.instanceId = instanceId;
+        return this;
+    }
+
+    /**
+     * Gets the custom instance ID for the sub-orchestration.
+     * @return The instance ID, or null if not set.
+     */
+    public String getInstanceId() {
+        return instanceId;
+    }
+}

client/src/main/java/com/microsoft/durabletask/OrchestrationRunner.java

@@ -129,7 +129,8 @@ public TaskOrchestration create() {
                 orchestrationFactories,
                 new JacksonDataConverter(),
                 DEFAULT_MAXIMUM_TIMER_INTERVAL,
-                logger);
+                logger,
+                null);
 
         // TODO: Error handling
         TaskOrchestratorResult taskOrchestratorResult = taskOrchestrationExecutor.execute(

client/src/main/java/com/microsoft/durabletask/TaskOptions.java

@@ -5,7 +5,7 @@
 /**
  * Options that can be used to control the behavior of orchestrator and activity task execution.
  */
-public final class TaskOptions {
+public class TaskOptions {
     private final RetryPolicy retryPolicy;
     private final RetryHandler retryHandler;
 

client/src/main/java/com/microsoft/durabletask/TaskOrchestrationExecutor.java

@@ -30,21 +30,29 @@ final class TaskOrchestrationExecutor {
     private final DataConverter dataConverter;
     private final Logger logger;
     private final Duration maximumTimerInterval;
+    private final DurableTaskGrpcWorkerVersioningOptions versioningOptions;
 
     public TaskOrchestrationExecutor(
             HashMap<String, TaskOrchestrationFactory> orchestrationFactories,
             DataConverter dataConverter,
             Duration maximumTimerInterval,
-            Logger logger) {
+            Logger logger,
+            DurableTaskGrpcWorkerVersioningOptions versioningOptions) {
         this.orchestrationFactories = orchestrationFactories;
         this.dataConverter = dataConverter;
         this.maximumTimerInterval = maximumTimerInterval;
         this.logger = logger;
+        this.versioningOptions = versioningOptions;
     }
 
     public TaskOrchestratorResult execute(List<HistoryEvent> pastEvents, List<HistoryEvent> newEvents) {
         ContextImplTask context = new ContextImplTask(pastEvents, newEvents);
 
+        if (this.versioningOptions != null && this.versioningOptions.getDefaultVersion() != null) {
+            // Set the default version for the orchestrator
+            context.setDefaultVersion(this.versioningOptions.getDefaultVersion());
+        }
+
         boolean completed = false;
         try {
             // Play through the history events until either we've played through everything
@@ -82,6 +90,7 @@ private class ContextImplTask implements TaskOrchestrationContext {
         private boolean isReplaying = true;
         private int newUUIDCounter;
         private String version;
+        private String defaultVersion;
 
         // LinkedHashMap to maintain insertion order when returning the list of pending actions
         private final LinkedHashMap<Integer, OrchestratorAction> pendingActions = new LinkedHashMap<>();
@@ -182,6 +191,15 @@ private void setVersion(String version) {
             this.version = version;
         }
 
+        private String getDefaultVersion() {
+            return this.defaultVersion;
+        }
+
+        private void setDefaultVersion(String defaultVersion) {
+            // This is used when starting sub-orchestrations
+            this.defaultVersion = defaultVersion;
+        }
+
         public <V> Task<V> completedTask(V value) {
             CompletableTask<V> task = new CompletableTask<>();
             task.complete(value);
@@ -389,6 +407,16 @@ public <V> Task<V> callSubOrchestrator(
             }
             createSubOrchestrationActionBuilder.setInstanceId(instanceId);
 
+            if (options instanceof NewSubOrchestrationInstanceOptions && ((NewSubOrchestrationInstanceOptions)options).getVersion() != null) {
+                NewSubOrchestrationInstanceOptions subOrchestrationOptions = (NewSubOrchestrationInstanceOptions) options;
+                if (subOrchestrationOptions.getVersion() != null) {
+                    createSubOrchestrationActionBuilder.setVersion(StringValue.of(subOrchestrationOptions.getVersion()));
+                }
+            } else if (this.getDefaultVersion() != null) {
+                // If the options are not of the correct type, we still allow the version to be set
+                createSubOrchestrationActionBuilder.setVersion(StringValue.of(this.getDefaultVersion()));
+            }
+
             TaskFactory<V> taskFactory = () -> {
                 int id = this.sequenceNumber++;
                 this.pendingActions.put(id, OrchestratorAction.newBuilder()

client/src/test/java/com/microsoft/durabletask/IntegrationTests.java

@@ -1742,4 +1742,61 @@ void orchestrationVersionStrictMatchStrategy(String orchestrationVersion) {
             throw new RuntimeException(e);
         }
     }
+
+    @ParameterizedTest
+    @ValueSource(strings = {"null", "", "0.9"})
+    public void versionsAppliedToSubOrchestration(String subOrchestrationVersion) {
+        final String orchestratorName = "VersionOrchestration";
+        final String subOrchestratorName = "subOrchestration";
+        final String activityName = "SayVersion";
+        final String defaultVersion = "1.0";
+        final DurableTaskGrpcWorkerVersioningOptions versioningOptions = new DurableTaskGrpcWorkerVersioningOptions(
+            "1.0", // Version of the worker.
+            defaultVersion, // Default version, used to assign versions to suborchestrations created by the worker.
+            DurableTaskGrpcWorkerVersioningOptions.VersionMatchStrategy.CURRENTOROLDER,
+            DurableTaskGrpcWorkerVersioningOptions.VersionFailureStrategy.FAIL);
+        DurableTaskGrpcWorker worker = this.createWorkerBuilder()
+            .addOrchestrator(orchestratorName, ctx -> {
+                final NewSubOrchestrationInstanceOptions options = new NewSubOrchestrationInstanceOptions((RetryPolicy) null);
+                if (!subOrchestrationVersion.equals("null")) {
+                    options.setVersion(subOrchestrationVersion);
+                }
+                String output = ctx.callSubOrchestrator(subOrchestratorName, null, null, options, String.class).await();
+                ctx.complete(output);
+            })
+            .addOrchestrator(subOrchestratorName, ctx -> {
+                String version = ctx.getVersion();
+                String output = ctx.callActivity(activityName, version, String.class).await();
+                ctx.complete(output);
+            })
+            .addActivity(activityName, ctx -> {
+                return String.format("Version: %s", ctx.getInput(String.class));
+            })
+            .useVersioning(versioningOptions)
+            .buildAndStart();
+
+        DurableTaskClient client = this.createClientBuilder()
+            .defaultVersion(defaultVersion)
+            .build();
+        try (worker; client) {
+            String instanceId = client.scheduleNewOrchestrationInstance(orchestratorName);
+            OrchestrationMetadata instance = client.waitForInstanceCompletion(
+                instanceId,
+                defaultTimeout,
+                true);
+
+            assertNotNull(instance);
+            assertEquals(OrchestrationRuntimeStatus.COMPLETED, instance.getRuntimeStatus());
+            String output = instance.readOutputAs(String.class);
+            String expected = "";
+            if (subOrchestrationVersion.equals("null")) {
+                expected = String.format("Version: %s", defaultVersion);
+            } else {
+                expected = String.format("Version: %s", subOrchestrationVersion);
+            }
+            assertEquals(expected, output);
+        } catch (TimeoutException e) {
+            throw new RuntimeException(e);
+        }
+    }
 }