Use dapr/durabletask-java (#1336)

* microsoft durabletask-java -> dapr durabletask-java

Signed-off-by: Cassandra Coyle <[email protected]>

* update another ref

Signed-off-by: Cassandra Coyle <[email protected]>

* 1.5.2 release

Signed-off-by: Cassandra Coyle <[email protected]>

* fix import order

Signed-off-by: Cassandra Coyle <[email protected]>

* Sdk new changes

Signed-off-by: siri-varma <[email protected]>

* Refine workflows

Signed-off-by: siri-varma <[email protected]>

* add ;

Signed-off-by: Cassandra Coyle <[email protected]>

* rm try

Signed-off-by: Cassandra Coyle <[email protected]>

---------

Signed-off-by: Cassandra Coyle <[email protected]>
Signed-off-by: siri-varma <[email protected]>
Co-authored-by: siri-varma <[email protected]>

Author: cicoyle

dapr-spring/dapr-spring-workflows/src/main/java/io/dapr/spring/workflows/config/DaprWorkflowsConfiguration.java

@@ -46,10 +46,9 @@ private void registerWorkflowsAndActivities(ApplicationContext applicationContex
       workflowRuntimeBuilder.registerActivity(activity);
     }
 
-    try (WorkflowRuntime runtime = workflowRuntimeBuilder.build()) {
-      LOGGER.info("Starting workflow runtime ... ");
-      runtime.start(false);
-    }
+    WorkflowRuntime runtime = workflowRuntimeBuilder.build();
+    LOGGER.info("Starting workflow runtime ... ");
+    runtime.start(false);
   }
 
   @Override

examples/src/main/java/io/dapr/examples/unittesting/DaprWorkflowExampleTest.java

@@ -13,8 +13,8 @@
 
 package io.dapr.examples.unittesting;
 
-import com.microsoft.durabletask.Task;
-import com.microsoft.durabletask.TaskCanceledException;
+import io.dapr.durabletask.Task;
+import io.dapr.durabletask.TaskCanceledException;
 import io.dapr.workflows.Workflow;
 import io.dapr.workflows.WorkflowContext;
 import io.dapr.workflows.WorkflowStub;

examples/src/main/java/io/dapr/examples/workflows/chain/DemoChainWorker.java

@@ -29,9 +29,8 @@ public static void main(String[] args) throws Exception {
     builder.registerActivity(ToUpperCaseActivity.class);
 
     // Build and then start the workflow runtime pulling and executing tasks
-    try (WorkflowRuntime runtime = builder.build()) {
-      System.out.println("Start workflow runtime");
-      runtime.start();
-    }
+    WorkflowRuntime runtime = builder.build();
+    System.out.println("Start workflow runtime");
+    runtime.start();
   }
 }

examples/src/main/java/io/dapr/examples/workflows/childworkflow/DemoChildWorkflowWorker.java

@@ -31,9 +31,7 @@ public static void main(String[] args) throws Exception {
     builder.registerActivity(ReverseActivity.class);
 
     // Build and then start the workflow runtime pulling and executing tasks
-    try (WorkflowRuntime runtime = builder.build()) {
-      System.out.println("Start workflow runtime");
-      runtime.start();
-    }
+    WorkflowRuntime runtime = builder.build();
+    System.out.println("Start workflow runtime");
   }
 }

examples/src/main/java/io/dapr/examples/workflows/continueasnew/DemoContinueAsNewWorker.java

@@ -16,6 +16,9 @@
 import io.dapr.workflows.runtime.WorkflowRuntime;
 import io.dapr.workflows.runtime.WorkflowRuntimeBuilder;
 
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
 public class DemoContinueAsNewWorker {
   /**
    * The main method of this app.
@@ -25,13 +28,14 @@ public class DemoContinueAsNewWorker {
    */
   public static void main(String[] args) throws Exception {
     // Register the Workflow with the builder.
-    WorkflowRuntimeBuilder builder = new WorkflowRuntimeBuilder().registerWorkflow(DemoContinueAsNewWorkflow.class);
+    WorkflowRuntimeBuilder builder = new WorkflowRuntimeBuilder().
+            registerWorkflow(DemoContinueAsNewWorkflow.class)
+            .withExecutorService(Executors.newFixedThreadPool(3));
     builder.registerActivity(CleanUpActivity.class);
 
     // Build and then start the workflow runtime pulling and executing tasks
-    try (WorkflowRuntime runtime = builder.build()) {
-      System.out.println("Start workflow runtime");
-      runtime.start();
-    }
+    WorkflowRuntime runtime = builder.build();
+    System.out.println("Start workflow runtime");
+    runtime.start();
   }
 }

examples/src/main/java/io/dapr/examples/workflows/externalevent/DemoExternalEventWorker.java

@@ -30,9 +30,8 @@ public static void main(String[] args) throws Exception {
     builder.registerActivity(DenyActivity.class);
 
     // Build and then start the workflow runtime pulling and executing tasks
-    try (WorkflowRuntime runtime = builder.build()) {
-      System.out.println("Start workflow runtime");
-      runtime.start();
-    }
+    WorkflowRuntime runtime = builder.build();
+    System.out.println("Start workflow runtime");
+    runtime.start();
   }
 }

examples/src/main/java/io/dapr/examples/workflows/faninout/DemoFanInOutWorker.java

@@ -29,9 +29,8 @@ public static void main(String[] args) throws Exception {
     builder.registerActivity(CountWordsActivity.class);
 
     // Build and then start the workflow runtime pulling and executing tasks
-    try (WorkflowRuntime runtime = builder.build()) {
-      System.out.println("Start workflow runtime");
-      runtime.start();
-    }
+    WorkflowRuntime runtime = builder.build();
+    System.out.println("Start workflow runtime");
+    runtime.start(false);
   }
 }

examples/src/main/java/io/dapr/examples/workflows/faninout/DemoFanInOutWorkflow.java

@@ -13,7 +13,7 @@
 
 package io.dapr.examples.workflows.faninout;
 
-import com.microsoft.durabletask.Task;
+import io.dapr.durabletask.Task;
 import io.dapr.workflows.Workflow;
 import io.dapr.workflows.WorkflowStub;
 

pom.xml

@@ -36,7 +36,7 @@
     <!--
         manually declare durabletask-client's jackson dependencies for workflows sdk
         which conflict with dapr-sdk's jackson dependencies
-        https://github.com/microsoft/durabletask-java/blob/main/client/build.gradle#L16
+        https://github.com/dapr/durabletask-java/blob/main/client/build.gradle#L16
         -->
     <jackson.version>2.16.1</jackson.version>
     <gpg.skip>true</gpg.skip>

sdk-tests/src/test/java/io/dapr/it/testcontainers/DaprWorkflowsIT.java

@@ -91,10 +91,9 @@ static void daprProperties(DynamicPropertyRegistry registry) {
    */
   @BeforeEach
   public void init() {
-    try (WorkflowRuntime runtime = workflowRuntimeBuilder.build()) {
-      System.out.println("Start workflow runtime");
-      runtime.start(false);
-    }
+    WorkflowRuntime runtime = workflowRuntimeBuilder.build();
+    System.out.println("Start workflow runtime");
+    runtime.start(false);
   }
 
   @Test

sdk-workflows/pom.xml

@@ -45,14 +45,14 @@
       <scope>test</scope>
     </dependency>
     <dependency>
-      <groupId>com.microsoft</groupId>
+      <groupId>io.dapr</groupId>
       <artifactId>durabletask-client</artifactId>
-      <version>1.5.0</version>
+      <version>1.5.2</version>
     </dependency>
     <!--
     manually declare durabletask-client's jackson dependencies
     which conflict with dapr-sdk's jackson dependencies
-    https://github.com/microsoft/durabletask-java/blob/main/client/build.gradle#L16
+    https://github.com/dapr/durabletask-java/blob/main/client/build.gradle#L16
     -->
     <dependency>
       <groupId>com.fasterxml.jackson.core</groupId>

sdk-workflows/src/main/java/io/dapr/workflows/WorkflowContext.java

@@ -13,10 +13,10 @@
 
 package io.dapr.workflows;
 
-import com.microsoft.durabletask.CompositeTaskFailedException;
-import com.microsoft.durabletask.Task;
-import com.microsoft.durabletask.TaskCanceledException;
-import com.microsoft.durabletask.TaskFailedException;
+import io.dapr.durabletask.CompositeTaskFailedException;
+import io.dapr.durabletask.Task;
+import io.dapr.durabletask.TaskCanceledException;
+import io.dapr.durabletask.TaskFailedException;
 import org.slf4j.Logger;
 
 import javax.annotation.Nullable;

sdk-workflows/src/main/java/io/dapr/workflows/client/DaprWorkflowClient.java

@@ -13,12 +13,12 @@
 
 package io.dapr.workflows.client;
 
-import com.microsoft.durabletask.DurableTaskClient;
-import com.microsoft.durabletask.DurableTaskGrpcClientBuilder;
-import com.microsoft.durabletask.NewOrchestrationInstanceOptions;
-import com.microsoft.durabletask.OrchestrationMetadata;
-import com.microsoft.durabletask.PurgeResult;
 import io.dapr.config.Properties;
+import io.dapr.durabletask.DurableTaskClient;
+import io.dapr.durabletask.DurableTaskGrpcClientBuilder;
+import io.dapr.durabletask.NewOrchestrationInstanceOptions;
+import io.dapr.durabletask.OrchestrationMetadata;
+import io.dapr.durabletask.PurgeResult;
 import io.dapr.utils.NetworkUtils;
 import io.dapr.workflows.Workflow;
 import io.dapr.workflows.internal.ApiTokenClientInterceptor;

sdk-workflows/src/main/java/io/dapr/workflows/runtime/DefaultWorkflowActivityContext.java

@@ -13,7 +13,7 @@
 
 package io.dapr.workflows.runtime;
 
-import com.microsoft.durabletask.TaskActivityContext;
+import io.dapr.durabletask.TaskActivityContext;
 import io.dapr.workflows.WorkflowActivityContext;
 
 /**

sdk-workflows/src/main/java/io/dapr/workflows/runtime/DefaultWorkflowContext.java

@@ -13,12 +13,12 @@
 
 package io.dapr.workflows.runtime;
 
-import com.microsoft.durabletask.CompositeTaskFailedException;
-import com.microsoft.durabletask.RetryPolicy;
-import com.microsoft.durabletask.Task;
-import com.microsoft.durabletask.TaskCanceledException;
-import com.microsoft.durabletask.TaskOptions;
-import com.microsoft.durabletask.TaskOrchestrationContext;
+import io.dapr.durabletask.CompositeTaskFailedException;
+import io.dapr.durabletask.RetryPolicy;
+import io.dapr.durabletask.Task;
+import io.dapr.durabletask.TaskCanceledException;
+import io.dapr.durabletask.TaskOptions;
+import io.dapr.durabletask.TaskOrchestrationContext;
 import io.dapr.workflows.WorkflowContext;
 import io.dapr.workflows.WorkflowTaskOptions;
 import io.dapr.workflows.WorkflowTaskRetryPolicy;

sdk-workflows/src/main/java/io/dapr/workflows/runtime/DefaultWorkflowFailureDetails.java

@@ -13,7 +13,7 @@
 
 package io.dapr.workflows.runtime;
 
-import com.microsoft.durabletask.FailureDetails;
+import io.dapr.durabletask.FailureDetails;
 import io.dapr.workflows.client.WorkflowFailureDetails;
 
 /**

sdk-workflows/src/main/java/io/dapr/workflows/runtime/DefaultWorkflowInstanceStatus.java

@@ -13,9 +13,9 @@
 
 package io.dapr.workflows.runtime;
 
-import com.microsoft.durabletask.FailureDetails;
-import com.microsoft.durabletask.OrchestrationMetadata;
-import com.microsoft.durabletask.OrchestrationRuntimeStatus;
+import io.dapr.durabletask.FailureDetails;
+import io.dapr.durabletask.OrchestrationMetadata;
+import io.dapr.durabletask.OrchestrationRuntimeStatus;
 import io.dapr.workflows.client.WorkflowFailureDetails;
 import io.dapr.workflows.client.WorkflowInstanceStatus;
 import io.dapr.workflows.client.WorkflowRuntimeStatus;

sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowActivityClassWrapper.java

@@ -13,8 +13,8 @@
 
 package io.dapr.workflows.runtime;
 
-import com.microsoft.durabletask.TaskActivity;
-import com.microsoft.durabletask.TaskActivityFactory;
+import io.dapr.durabletask.TaskActivity;
+import io.dapr.durabletask.TaskActivityFactory;
 import io.dapr.workflows.WorkflowActivity;
 
 import java.lang.reflect.Constructor;

sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowActivityInstanceWrapper.java

@@ -13,8 +13,8 @@
 
 package io.dapr.workflows.runtime;
 
-import com.microsoft.durabletask.TaskActivity;
-import com.microsoft.durabletask.TaskActivityFactory;
+import io.dapr.durabletask.TaskActivity;
+import io.dapr.durabletask.TaskActivityFactory;
 import io.dapr.workflows.WorkflowActivity;
 
 /**

sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowClassWrapper.java

@@ -13,8 +13,8 @@
 
 package io.dapr.workflows.runtime;
 
-import com.microsoft.durabletask.TaskOrchestration;
-import com.microsoft.durabletask.TaskOrchestrationFactory;
+import io.dapr.durabletask.TaskOrchestration;
+import io.dapr.durabletask.TaskOrchestrationFactory;
 import io.dapr.workflows.Workflow;
 
 import java.lang.reflect.Constructor;

sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowInstanceWrapper.java

@@ -13,8 +13,8 @@
 
 package io.dapr.workflows.runtime;
 
-import com.microsoft.durabletask.TaskOrchestration;
-import com.microsoft.durabletask.TaskOrchestrationFactory;
+import io.dapr.durabletask.TaskOrchestration;
+import io.dapr.durabletask.TaskOrchestrationFactory;
 import io.dapr.workflows.Workflow;
 
 /**

sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowRuntime.java

@@ -13,17 +13,34 @@
 
 package io.dapr.workflows.runtime;
 
-import com.microsoft.durabletask.DurableTaskGrpcWorker;
+import io.dapr.durabletask.DurableTaskGrpcWorker;
+import io.grpc.ManagedChannel;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
 
 /**
  * Contains methods to register workflows and activities.
  */
 public class WorkflowRuntime implements AutoCloseable {
 
-  private DurableTaskGrpcWorker worker;
+  private final DurableTaskGrpcWorker worker;
+  private final ManagedChannel managedChannel;
+  private final ExecutorService executorService;
 
-  public WorkflowRuntime(DurableTaskGrpcWorker worker) {
+  /**
+   * Constructor.
+   *
+   * @param worker grpcWorker processing activities.
+   * @param managedChannel grpc channel.
+   * @param executorService executor service responsible for running the threads.
+   */
+  public WorkflowRuntime(DurableTaskGrpcWorker worker,
+                         ManagedChannel managedChannel,
+                         ExecutorService executorService) {
     this.worker = worker;
+    this.managedChannel = managedChannel;
+    this.executorService = executorService;
   }
 
   /**
@@ -50,11 +67,31 @@ public void start(boolean block) {
   /**
    * {@inheritDoc}
    */
-  @Override
   public void close() {
-    if (this.worker != null) {
-      this.worker.close();
-      this.worker = null;
+    this.shutDownWorkerPool();
+    this.closeSideCarChannel();
+  }
+
+  private void closeSideCarChannel() {
+    this.managedChannel.shutdown();
+
+    try {
+      if (!this.managedChannel.awaitTermination(60, TimeUnit.SECONDS)) {
+        this.managedChannel.shutdownNow();
+      }
+    } catch (InterruptedException ex) {
+      Thread.currentThread().interrupt();
+    }
+  }
+
+  private void shutDownWorkerPool() {
+    this.executorService.shutdown();
+    try {
+      if (!this.executorService.awaitTermination(60, TimeUnit.SECONDS)) {
+        this.executorService.shutdownNow();
+      }
+    } catch (InterruptedException ex) {
+      Thread.currentThread().interrupt();
     }
   }
 }