chore: Propagate taskExecutionId from runtime to activityContext

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

Author: javier-aliaga

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

@@ -185,7 +185,7 @@ public void startAndBlock() {
                                 output = taskActivityExecutor.execute(
                                         activityRequest.getName(),
                                         activityRequest.getInput().getValue(),
-                                        activityRequest.getTaskId());
+                                        activityRequest.getTaskExecutionId());
                             } catch (Throwable e) {
                                 failureDetails = TaskFailureDetails.newBuilder()
                                         .setErrorType(e.getClass().getName())

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

@@ -6,9 +6,8 @@
 
 import java.time.Duration;
 import java.util.HashMap;
-import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
+
 
 /**
  * Builder object for constructing customized {@link DurableTaskGrpcWorker} instances.

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

@@ -21,4 +21,11 @@ public interface TaskActivityContext {
      * @return the deserialized activity input value
      */
     <T> T getInput(Class<T> targetType);
+
+
+    /**
+     * Gets the execution id of the current task activity.
+     * @return the execution id of the current task activity
+     */
+    String getTaskExecutionId();
 }

client/src/main/java/io/dapr/durabletask/TaskActivityExecutor.java

@@ -19,7 +19,7 @@ public TaskActivityExecutor(
         this.logger = logger;
     }
 
-    public String execute(String taskName, String input, int taskId) throws Throwable {
+    public String execute(String taskName, String input, String taskExecutionId) throws Throwable {
         TaskActivityFactory factory = this.activityFactories.get(taskName);
         if (factory == null) {
             throw new IllegalStateException(
@@ -32,7 +32,7 @@ public String execute(String taskName, String input, int taskId) throws Throwabl
                     String.format("The task factory '%s' returned a null TaskActivity object.", taskName));
         }
 
-        TaskActivityContextImpl context = new TaskActivityContextImpl(taskName, input);
+        TaskActivityContextImpl context = new TaskActivityContextImpl(taskName, input, taskExecutionId);
 
         // Unhandled exceptions are allowed to escape
         Object output = activity.run(context);
@@ -46,12 +46,14 @@ public String execute(String taskName, String input, int taskId) throws Throwabl
     private class TaskActivityContextImpl implements TaskActivityContext {
         private final String name;
         private final String rawInput;
+        private final String taskExecutionId;
 
         private final DataConverter dataConverter = TaskActivityExecutor.this.dataConverter;
 
-        public TaskActivityContextImpl(String activityName, String rawInput) {
+        public TaskActivityContextImpl(String activityName, String rawInput, String taskExecutionId) {
             this.name = activityName;
             this.rawInput = rawInput;
+            this.taskExecutionId = taskExecutionId;
         }
 
         @Override
@@ -67,5 +69,10 @@ public <T> T getInput(Class<T> targetType) {
 
             return this.dataConverter.deserialize(this.rawInput, targetType);
         }
+
+        @Override
+        public String getTaskExecutionId() {
+            return this.taskExecutionId;
+        }
     }
 }

client/src/main/java/io/dapr/durabletask/TaskOrchestrationExecutor.java

@@ -265,7 +265,7 @@ public <V> Task<V> callActivity(
             }
 
             String serializedInput = this.dataConverter.serialize(input);
-            Builder scheduleTaskBuilder = ScheduleTaskAction.newBuilder().setName(name);
+            Builder scheduleTaskBuilder = ScheduleTaskAction.newBuilder().setName(name).setTaskExecutionId(newUUID().toString());
             if (serializedInput != null) {
                 scheduleTaskBuilder.setInput(StringValue.of(serializedInput));
             }

client/src/test/java/io/dapr/durabletask/IntegrationTests.java

@@ -1566,7 +1566,10 @@ public void newUUIDTest() {
                     if (currentUUID1.equals(currentUUID2)) ctx.complete(false);
                     else ctx.complete(true);
                 })
-                .addActivity(echoActivityName, ctx -> ctx.getInput(UUID.class))
+                .addActivity(echoActivityName, ctx -> {
+                    System.out.println("##### echoActivityName: " + ctx.getInput(UUID.class ));
+                    return ctx.getInput(UUID.class);
+                })
                 .buildAndStart();
         DurableTaskClient client = new DurableTaskGrpcClientBuilder().build();
 
@@ -1581,4 +1584,55 @@ public void newUUIDTest() {
         }
 
     }
-}
+
+
+    @Test
+    public void taskExecutionIdTest() {
+        var orchestratorName = "test-task-execution-id";
+        var retryActivityName = "RetryN";
+        final RetryPolicy retryPolicy = new RetryPolicy(4, Duration.ofSeconds(3));
+        final TaskOptions taskOptions = new TaskOptions(retryPolicy);
+
+        var execMap = new HashMap<String, Integer>();
+
+        DurableTaskGrpcWorker worker = this.createWorkerBuilder()
+                .addOrchestrator(orchestratorName, ctx -> {
+                    ctx.callActivity(retryActivityName,null,taskOptions).await();    
+                    ctx.callActivity(retryActivityName,null,taskOptions).await();    
+                    ctx.complete(true);
+                })
+                .addActivity(retryActivityName, ctx -> {
+                    System.out.println("##### RetryN[executionId]: " + ctx.getTaskExecutionId());
+                    var c = execMap.get(ctx.getTaskExecutionId());
+                    if (c == null) {
+                        c = 0;
+                    } else {
+                        c++;
+                    }
+
+                    execMap.put(ctx.getTaskExecutionId(), c);
+                    if (c < 2) {
+                        throw new RuntimeException("test retry");
+                    }
+                    return null;
+                })
+                .buildAndStart();
+        DurableTaskClient client = new DurableTaskGrpcClientBuilder().build();
+
+        try(worker; client) {
+            client.createTaskHub(true);
+            String instanceId = client.scheduleNewOrchestrationInstance(orchestratorName);
+            OrchestrationMetadata instance = client.waitForInstanceCompletion(instanceId, defaultTimeout, true);
+            assertNotNull(instance);
+            assertEquals(OrchestrationRuntimeStatus.COMPLETED, instance.getRuntimeStatus());
+            assertEquals(2, execMap.size());
+            assertTrue(instance.readOutputAs(boolean.class));
+        } catch (TimeoutException e) {
+            throw new RuntimeException(e);
+        }
+
+    }
+
+}
+
+