feat: Add workflow versioning (dapr#1624)

Signed-off-by: Javier Aliaga <[email protected]>
Signed-off-by: salaboy <[email protected]>

Author: javier-aliaga

durabletask-client/pom.xml

@@ -59,6 +59,10 @@
       <groupId>com.fasterxml.jackson.datatype</groupId>
       <artifactId>jackson-datatype-jsr310</artifactId>
     </dependency>
+    <dependency>
+      <groupId>org.apache.commons</groupId>
+      <artifactId>commons-lang3</artifactId>
+    </dependency>
     <dependency>
       <groupId>io.grpc</groupId>
       <artifactId>grpc-testing</artifactId>

durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskClient.java

@@ -145,7 +145,7 @@ public void raiseEvent(String instanceId, String eventName) {
    * Waits for an orchestration to start running and returns an {@link OrchestrationMetadata} object that contains
    * metadata about the started instance.
    *
-   * <p> A "started" orchestration instance is any instance not in the <code>Pending</code> state. </p>
+   * <p>A "started" orchestration instance is any instance not in the <code>Pending</code> state. </p>
    *
    * <p>If an orchestration instance is already running when this method is called, the method will return immediately.
    *</p>

durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcWorker.java

@@ -17,11 +17,13 @@
 import io.dapr.durabletask.implementation.protobuf.OrchestratorService;
 import io.dapr.durabletask.implementation.protobuf.OrchestratorService.TaskFailureDetails;
 import io.dapr.durabletask.implementation.protobuf.TaskHubSidecarServiceGrpc;
+import io.dapr.durabletask.orchestration.TaskOrchestrationFactories;
 import io.grpc.Channel;
 import io.grpc.ManagedChannel;
 import io.grpc.ManagedChannelBuilder;
 import io.grpc.Status;
 import io.grpc.StatusRuntimeException;
+import org.apache.commons.lang3.StringUtils;
 
 import java.time.Duration;
 import java.util.HashMap;
@@ -42,7 +44,8 @@ public final class DurableTaskGrpcWorker implements AutoCloseable {
   private static final Logger logger = Logger.getLogger(DurableTaskGrpcWorker.class.getPackage().getName());
   private static final Duration DEFAULT_MAXIMUM_TIMER_INTERVAL = Duration.ofDays(3);
 
-  private final HashMap<String, TaskOrchestrationFactory> orchestrationFactories = new HashMap<>();
+  private final TaskOrchestrationFactories orchestrationFactories;
+
   private final HashMap<String, TaskActivityFactory> activityFactories = new HashMap<>();
 
   private final ManagedChannel managedSidecarChannel;
@@ -57,7 +60,7 @@ public final class DurableTaskGrpcWorker implements AutoCloseable {
   private Thread workerThread;
 
   DurableTaskGrpcWorker(DurableTaskGrpcWorkerBuilder builder) {
-    this.orchestrationFactories.putAll(builder.orchestrationFactories);
+    this.orchestrationFactories = builder.orchestrationFactories;
     this.activityFactories.putAll(builder.activityFactories);
     this.appId = builder.appId;
 
@@ -115,7 +118,9 @@ public void start() {
    *
    */
   public void close() {
-    this.workerThread.interrupt();
+    if (this.workerThread != null) {
+      this.workerThread.interrupt();
+    }
     this.isNormalShutdown = true;
     this.shutDownWorkerPool();
     this.closeSideCarChannel();
@@ -161,6 +166,7 @@ public void startAndBlock() {
           OrchestratorService.WorkItem.RequestCase requestType = workItem.getRequestCase();
           if (requestType == OrchestratorService.WorkItem.RequestCase.ORCHESTRATORREQUEST) {
             OrchestratorService.OrchestratorRequest orchestratorRequest = workItem.getOrchestratorRequest();
+
             logger.log(Level.FINEST,
                 String.format("Processing orchestrator request for instance: {0}",
                     orchestratorRequest.getInstanceId()));
@@ -171,11 +177,22 @@ public void startAndBlock() {
                   orchestratorRequest.getPastEventsList(),
                   orchestratorRequest.getNewEventsList());
 
+              var versionBuilder = OrchestratorService.OrchestrationVersion.newBuilder();
+
+              if (StringUtils.isNotEmpty(taskOrchestratorResult.getVersion())) {
+                versionBuilder.setName(taskOrchestratorResult.getVersion());
+              }
+
+              if (taskOrchestratorResult.getPatches() != null) {
+                versionBuilder.addAllPatches(taskOrchestratorResult.getPatches());
+              }
+
               OrchestratorService.OrchestratorResponse response = OrchestratorService.OrchestratorResponse.newBuilder()
                   .setInstanceId(orchestratorRequest.getInstanceId())
                   .addAllActions(taskOrchestratorResult.getActions())
                   .setCustomStatus(StringValue.of(taskOrchestratorResult.getCustomStatus()))
                   .setCompletionToken(workItem.getCompletionToken())
+                  .setVersion(versionBuilder)
                   .build();
 
               try {

durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcWorkerBuilder.java

@@ -13,6 +13,8 @@
 
 package io.dapr.durabletask;
 
+import io.dapr.durabletask.orchestration.TaskOrchestrationFactories;
+import io.dapr.durabletask.orchestration.TaskOrchestrationFactory;
 import io.grpc.Channel;
 
 import java.time.Duration;
@@ -24,7 +26,7 @@
  *
  */
 public final class DurableTaskGrpcWorkerBuilder {
-  final HashMap<String, TaskOrchestrationFactory> orchestrationFactories = new HashMap<>();
+  TaskOrchestrationFactories orchestrationFactories = new TaskOrchestrationFactories();
   final HashMap<String, TaskActivityFactory> activityFactories = new HashMap<>();
   int port;
   Channel channel;
@@ -40,17 +42,7 @@ public final class DurableTaskGrpcWorkerBuilder {
    * @return this builder object
    */
   public DurableTaskGrpcWorkerBuilder addOrchestration(TaskOrchestrationFactory factory) {
-    String key = factory.getName();
-    if (key == null || key.length() == 0) {
-      throw new IllegalArgumentException("A non-empty task orchestration name is required.");
-    }
-
-    if (this.orchestrationFactories.containsKey(key)) {
-      throw new IllegalArgumentException(
-          String.format("A task orchestration factory named %s is already registered.", key));
-    }
-
-    this.orchestrationFactories.put(key, factory);
+    this.orchestrationFactories.addOrchestration(factory);
     return this;
   }
 

durabletask-client/src/main/java/io/dapr/durabletask/OrchestrationMetadata.java

@@ -226,7 +226,7 @@ public boolean isCustomStatusFetched() {
   private <T> T readPayloadAs(Class<T> type, String payload) {
     if (!this.requestedInputsAndOutputs) {
       throw new IllegalStateException("This method can only be used when instance metadata is fetched with the option "
-         + "to include input and output data.");
+          + "to include input and output data.");
     }
 
     // Note that the Java gRPC implementation converts null protobuf strings into empty Java strings

durabletask-client/src/main/java/io/dapr/durabletask/OrchestrationRunner.java

@@ -16,10 +16,11 @@
 import com.google.protobuf.InvalidProtocolBufferException;
 import com.google.protobuf.StringValue;
 import io.dapr.durabletask.implementation.protobuf.OrchestratorService;
+import io.dapr.durabletask.orchestration.TaskOrchestrationFactories;
+import io.dapr.durabletask.orchestration.TaskOrchestrationFactory;
 
 import java.time.Duration;
 import java.util.Base64;
-import java.util.HashMap;
 import java.util.logging.Logger;
 
 /**
@@ -134,8 +135,8 @@ public static byte[] loadAndRun(byte[] orchestratorRequestBytes, TaskOrchestrati
     }
 
     // Register the passed orchestration as the default ("*") orchestration
-    HashMap<String, TaskOrchestrationFactory> orchestrationFactories = new HashMap<>();
-    orchestrationFactories.put("*", new TaskOrchestrationFactory() {
+    TaskOrchestrationFactories orchestrationFactories = new TaskOrchestrationFactories();
+    orchestrationFactories.addOrchestration(new TaskOrchestrationFactory() {
       @Override
       public String getName() {
         return "*";
@@ -145,6 +146,16 @@ public String getName() {
       public TaskOrchestration create() {
         return orchestration;
       }
+
+      @Override
+      public String getVersionName() {
+        return "";
+      }
+
+      @Override
+      public Boolean isLatestVersion() {
+        return false;
+      }
     });
 
     TaskOrchestrationExecutor taskOrchestrationExecutor = new TaskOrchestrationExecutor(

durabletask-client/src/main/java/io/dapr/durabletask/OrchestrationRuntimeStatus.java

@@ -68,7 +68,12 @@ public enum OrchestrationRuntimeStatus {
   /**
    * The orchestration is in a suspended state.
    */
-  SUSPENDED;
+  SUSPENDED,
+
+  /**
+   * The orchestration is in a stalled state.
+   */
+  STALLED;
 
   static OrchestrationRuntimeStatus fromProtobuf(OrchestratorService.OrchestrationStatus status) {
     switch (status) {
@@ -88,6 +93,8 @@ static OrchestrationRuntimeStatus fromProtobuf(OrchestratorService.Orchestration
         return PENDING;
       case ORCHESTRATION_STATUS_SUSPENDED:
         return SUSPENDED;
+      case ORCHESTRATION_STATUS_STALLED:
+        return STALLED;
       default:
         throw new IllegalArgumentException(String.format("Unknown status value: %s", status));
     }

durabletask-client/src/main/java/io/dapr/durabletask/Task.java

@@ -27,12 +27,14 @@
  * <pre>
  * Task{@literal <}int{@literal >} activityTask = ctx.callActivity("MyActivity", someInput, int.class);
  * </pre>
+ *
  * <p>Orchestrator code uses the {@link #await()} method to block on the completion of the task and retrieve the result.
  * If the task is not yet complete, the {@code await()} method will throw an {@link OrchestratorBlockedException}, which
  * pauses the orchestrator's execution so that it can save its progress into durable storage and schedule any
  * outstanding work. When the task is complete, the orchestrator will run again from the beginning and the next time
  * the task's {@code await()} method is called, the result will be returned, or a {@link TaskFailedException} will be
  * thrown if the result of the task was an unhandled exception.</p>
+ *
  * <p>Note that orchestrator code must never catch {@code OrchestratorBlockedException} because doing so can cause the
  * orchestration instance to get permanently stuck.</p>
  *

durabletask-client/src/main/java/io/dapr/durabletask/TaskActivityContext.java

@@ -34,7 +34,6 @@ public interface TaskActivityContext {
    */
   <T> T getInput(Class<T> targetType);
 
-
   /**
    * Gets the execution id of the current task activity.
    *

durabletask-client/src/main/java/io/dapr/durabletask/TaskFailedException.java

@@ -16,6 +16,7 @@
 /**
  * Exception that gets thrown when awaiting a {@link Task} for an activity or sub-orchestration that fails with an
  * unhandled exception.
+ *
  * <p>Detailed information associated with a particular task failure can be retrieved
  * using the {@link #getErrorDetails()} method.</p>
  */