From 77c49761ae0186db619aafe508e10d662edb3f11 Mon Sep 17 00:00:00 2001
From: salaboy <Salaboy@gmail.com>
Date: Tue, 5 Aug 2025 08:50:11 +0100
Subject: [PATCH] initial work

---
 sdk-workflows/pom.xml                         |  6 +++-
 .../workflows/client/DaprWorkflowClient.java  | 30 ++++++++++++++++---
 .../runtime/WorkflowRuntimeBuilder.java       |  8 +++--
 3 files changed, 37 insertions(+), 7 deletions(-)

diff --git a/sdk-workflows/pom.xml b/sdk-workflows/pom.xml
index 883489fd0..2f188c020 100644
--- a/sdk-workflows/pom.xml
+++ b/sdk-workflows/pom.xml
@@ -47,7 +47,11 @@
     <dependency>
       <groupId>io.dapr</groupId>
       <artifactId>durabletask-client</artifactId>
-      <version>1.5.6</version>
+      <version>1.5.8-SNAPSHOT</version>
+    </dependency>
+    <dependency>
+      <groupId>io.opentelemetry</groupId>
+      <artifactId>opentelemetry-api</artifactId>
     </dependency>
     <!--
     manually declare durabletask-client's jackson dependencies
diff --git a/sdk-workflows/src/main/java/io/dapr/workflows/client/DaprWorkflowClient.java b/sdk-workflows/src/main/java/io/dapr/workflows/client/DaprWorkflowClient.java
index d8b94edbe..7e3be0d8a 100644
--- a/sdk-workflows/src/main/java/io/dapr/workflows/client/DaprWorkflowClient.java
+++ b/sdk-workflows/src/main/java/io/dapr/workflows/client/DaprWorkflowClient.java
@@ -21,10 +21,9 @@
 import io.dapr.durabletask.PurgeResult;
 import io.dapr.utils.NetworkUtils;
 import io.dapr.workflows.Workflow;
-import io.dapr.workflows.internal.ApiTokenClientInterceptor;
 import io.dapr.workflows.runtime.DefaultWorkflowInstanceStatus;
-import io.grpc.ClientInterceptor;
 import io.grpc.ManagedChannel;
+import io.opentelemetry.context.Context;
 
 import javax.annotation.Nullable;
 
@@ -37,7 +36,6 @@
  */
 public class DaprWorkflowClient implements AutoCloseable {
 
-  private ClientInterceptor workflowApiTokenInterceptor;
   private DurableTaskClient innerClient;
   private ManagedChannel grpcChannel;
 
@@ -54,7 +52,7 @@ public DaprWorkflowClient() {
    * @param properties Properties for the GRPC Channel.
    */
   public DaprWorkflowClient(Properties properties) {
-    this(NetworkUtils.buildGrpcManagedChannel(properties, new ApiTokenClientInterceptor(properties)));
+    this(NetworkUtils.buildGrpcManagedChannel(properties));
   }
 
   /**
@@ -100,6 +98,11 @@ public <T extends Workflow> String scheduleNewWorkflow(Class<T> clazz, Object in
     return this.innerClient.scheduleNewOrchestrationInstance(clazz.getCanonicalName(), input);
   }
 
+  public <T extends Workflow> String scheduleNewWorkflow(Class<T> clazz, Object input, String instanceId,
+                                                         Context context) {
+    return this.innerClient.scheduleNewOrchestrationInstance(clazz.getCanonicalName(), input, instanceId, context);
+  }
+
   /**
    * Schedules a new workflow using DurableTask client.
    *
@@ -128,6 +131,23 @@ public <T extends Workflow> String scheduleNewWorkflow(Class<T> clazz, NewWorkfl
         orchestrationInstanceOptions);
   }
 
+  /**
+   * Schedules a new workflow with a specified set of options for execution.
+   *
+   * @param <T>        any Workflow type
+   * @param clazz      Class extending Workflow to start an instance of.
+   * @param options the options for the new workflow, including input, instance ID, etc.
+   * @param context otel Context for trace propagation.
+   * @return the <code>instanceId</code> parameter value.
+   */
+  public <T extends Workflow> String scheduleNewWorkflow(Class<T> clazz, NewWorkflowOptions options,
+                                                         Context context) {
+    NewOrchestrationInstanceOptions orchestrationInstanceOptions = fromNewWorkflowOptions(options);
+
+    return this.innerClient.scheduleNewOrchestrationInstance(clazz.getCanonicalName(),
+            orchestrationInstanceOptions, context);
+  }
+
   /**
    * Suspend the workflow associated with the provided instance id.
    *
@@ -278,8 +298,10 @@ public void close() throws InterruptedException {
    * @return a new instance of a DurableTaskClient with a GRPC channel.
    */
   private static DurableTaskClient createDurableTaskClient(ManagedChannel grpcChannel) {
+
     return new DurableTaskGrpcClientBuilder()
         .grpcChannel(grpcChannel)
+            .interceptor(new io.dapr.durabletask.interceptors.DaprWorkflowClientGrpcInterceptors())
         .build();
   }
 
diff --git a/sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowRuntimeBuilder.java b/sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowRuntimeBuilder.java
index 5daff7747..82c6e5782 100644
--- a/sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowRuntimeBuilder.java
+++ b/sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowRuntimeBuilder.java
@@ -15,12 +15,14 @@
 
 import io.dapr.config.Properties;
 import io.dapr.durabletask.DurableTaskGrpcWorkerBuilder;
+import io.dapr.durabletask.interceptors.DaprWorkflowClientGrpcInterceptors;
 import io.dapr.utils.NetworkUtils;
 import io.dapr.workflows.Workflow;
 import io.dapr.workflows.WorkflowActivity;
 import io.dapr.workflows.internal.ApiTokenClientInterceptor;
 import io.grpc.ClientInterceptor;
 import io.grpc.ManagedChannel;
+import io.opentelemetry.context.Context;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -65,7 +67,8 @@ public WorkflowRuntimeBuilder(Logger logger) {
   private WorkflowRuntimeBuilder(Properties properties, Logger logger) {
     this.workflowApiTokenInterceptor = new ApiTokenClientInterceptor(properties);
     this.managedChannel = NetworkUtils.buildGrpcManagedChannel(properties, workflowApiTokenInterceptor);
-    this.builder = new DurableTaskGrpcWorkerBuilder().grpcChannel(this.managedChannel);
+    this.builder = new DurableTaskGrpcWorkerBuilder().grpcChannel(this.managedChannel)
+            .interceptors(new DaprWorkflowClientGrpcInterceptors());
     this.logger = logger;
   }
 
@@ -77,7 +80,8 @@ private WorkflowRuntimeBuilder(Properties properties, Logger logger) {
   public WorkflowRuntime build() {
     if (instance == null) {
       synchronized (WorkflowRuntime.class) {
-        this.executorService = this.executorService == null ? Executors.newCachedThreadPool() : this.executorService;
+        this.executorService = Context.taskWrapping(this.executorService == null ? Executors.newCachedThreadPool()
+                : this.executorService);
         if (instance == null) {
           instance = new WorkflowRuntime(
                   this.builder.withExecutorService(this.executorService).build(),