chore: New task execution task id test (#1352)

* chore: New task execution task id test

test how taskExecutionTaskId can be used for idempotency

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

* chore: Clean up not used files

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

* docs: Task execution keys

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

* test: Modify unit tests

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

* Remove new lines

Signed-off-by: artur-ciocanu <[email protected]>

---------

Signed-off-by: Javier Aliaga <[email protected]>
Signed-off-by: artur-ciocanu <[email protected]>
Co-authored-by: Cassie Coyle <[email protected]>
Co-authored-by: artur-ciocanu <[email protected]>
Signed-off-by: Javier Aliaga <[email protected]>

Author: 3 people

daprdocs/content/en/java-sdk-docs/java-workflow/java-workflow-howto.md

@@ -6,7 +6,7 @@ weight: 20000
 description: How to get up and running with workflows using the Dapr Java SDK
 ---
 
-Let’s create a Dapr workflow and invoke it using the console. With the [provided workflow example](https://github.com/dapr/java-sdk/tree/master/examples/src/main/java/io/dapr/examples/workflows), you will:
+Let's create a Dapr workflow and invoke it using the console. With the [provided workflow example](https://github.com/dapr/java-sdk/tree/master/examples/src/main/java/io/dapr/examples/workflows), you will:
 
 - Execute the workflow instance using the [Java workflow worker](https://github.com/dapr/java-sdk/blob/master/examples/src/main/java/io/dapr/examples/workflows/DemoWorkflowWorker.java)
 - Utilize the Java workflow client and API calls to [start and terminate workflow instances](https://github.com/dapr/java-sdk/blob/master/examples/src/main/java/io/dapr/examples/workflows/DemoWorkflowClient.java)
@@ -85,11 +85,10 @@ You're up and running! Both Dapr and your app logs will appear here.
 == APP == INFO: Durable Task worker is connecting to sidecar at 127.0.0.1:50001.
 ```
 
-## Run the `DemoWorkflowClient
+## Run the `DemoWorkflowClient`
 
 The `DemoWorkflowClient` starts instances of workflows that have been registered with Dapr.
 
-
 ```java
 public class DemoWorkflowClient {
 
@@ -246,4 +245,40 @@ Exiting DemoWorkflowClient.
 
 ## Next steps
 - [Learn more about Dapr workflow]({{% ref workflow-overview.md %}})
-- [Workflow API reference]({{% ref workflow_api.md %}})
+- [Workflow API reference]({{% ref workflow_api.md %}})
+
+## Advanced features
+
+### Task Execution Keys
+
+Task execution keys are unique identifiers generated by the durabletask-java library. They are stored in the `WorkflowActivityContext` and can be used to track and manage the execution of workflow activities. They are particularly useful for:
+
+1. **Idempotency**: Ensuring activities are not executed multiple times for the same task
+2. **State Management**: Tracking the state of activity execution
+3. **Error Handling**: Managing retries and failures in a controlled manner
+
+Here's an example of how to use task execution keys in your workflow activities:
+
+```java
+public class TaskExecutionKeyActivity implements WorkflowActivity {
+  @Override
+  public Object run(WorkflowActivityContext ctx) {
+    // Get the task execution key for this activity
+    String taskExecutionKey = ctx.getTaskExecutionKey();
+    
+    // Use the key to implement idempotency or state management
+    // For example, check if this task has already been executed
+    if (isTaskAlreadyExecuted(taskExecutionKey)) {
+      return getPreviousResult(taskExecutionKey);
+    }
+    
+    // Execute the activity logic
+    Object result = executeActivityLogic();
+    
+    // Store the result with the task execution key
+    storeResult(taskExecutionKey, result);
+    
+    return result;
+  }
+}
+```

pom.xml

@@ -49,7 +49,7 @@
     <snakeyaml.version>2.0</snakeyaml.version>
     <testcontainers.version>1.21.3</testcontainers.version>
     <!-- Do NOT UPGRADE spring.version without checking springboot.version alignment -->
-    <springboot.version>3.4.6</springboot.version>
+    <springboot.version>3.4.9</springboot.version>
     <springframework.version>6.2.7</springframework.version>
     <!-- Do NOT UPGRADE springframework.version without checking springboot.version alignment -->
     <nexus-staging-maven-plugin.version>1.7.0</nexus-staging-maven-plugin.version>
@@ -65,6 +65,8 @@
     <commons-io.version>2.14.0</commons-io.version>
     <zipkin.version>3.4.0</zipkin.version>
     <microcks.version>0.3.1</microcks.version>
+    <commons-compress.version>1.26.0</commons-compress.version>
+    <commons-codec.version>1.17.0</commons-codec.version>
   </properties>
 
   <distributionManagement>
@@ -372,6 +374,17 @@
         <artifactId>wiremock-standalone</artifactId>
         <version>${wiremock.version}</version>
       </dependency>
+      <dependency>
+        <groupId>org.apache.commons</groupId>
+        <artifactId>commons-compress</artifactId>
+        <version>${commons-compress.version}</version>
+      </dependency>
+      <dependency>
+        <groupId>commons-codec</groupId>
+        <artifactId>commons-codec</artifactId>
+        <version>${commons-codec.version}</version>
+        <scope>testf</scope>
+      </dependency>
     </dependencies>
   </dependencyManagement>
 

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

@@ -15,6 +15,7 @@
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
+
 import io.dapr.testcontainers.Component;
 import io.dapr.testcontainers.DaprContainer;
 import io.dapr.testcontainers.DaprLogLevel;
@@ -41,6 +42,7 @@
 import java.util.Map;
 
 import static io.dapr.it.testcontainers.ContainerConstants.DAPR_RUNTIME_IMAGE_TAG;
+import static org.junit.Assert.assertTrue;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 
@@ -153,7 +155,7 @@ public void testNamedActivitiesWorkflows() throws Exception {
     String instanceId = workflowClient.scheduleNewWorkflow(TestNamedActivitiesWorkflow.class, payload);
 
     workflowClient.waitForInstanceStart(instanceId, Duration.ofSeconds(10), false);
-    
+
     Duration timeout = Duration.ofSeconds(10);
     WorkflowInstanceStatus workflowStatus = workflowClient.waitForInstanceCompletion(instanceId, timeout, true);
 
@@ -171,6 +173,28 @@ public void testNamedActivitiesWorkflows() throws Exception {
     assertEquals(instanceId, workflowOutput.getWorkflowId());
   }
 
+  @Test
+  public void testExecutionKeyWorkflows() throws Exception {
+    TestWorkflowPayload payload = new TestWorkflowPayload(new ArrayList<>());
+    String instanceId = workflowClient.scheduleNewWorkflow(TestExecutionKeysWorkflow.class, payload);
+
+    workflowClient.waitForInstanceStart(instanceId, Duration.ofSeconds(100), false);
+    
+    Duration timeout = Duration.ofSeconds(1000);
+    WorkflowInstanceStatus workflowStatus = workflowClient.waitForInstanceCompletion(instanceId, timeout, true);
+
+    assertNotNull(workflowStatus);
+
+    TestWorkflowPayload workflowOutput = deserialize(workflowStatus.getSerializedOutput());
+
+    assertEquals(1, workflowOutput.getPayloads().size());
+    assertEquals("Execution key found", workflowOutput.getPayloads().get(0));
+    
+    assertTrue(KeyStore.getInstance().size() == 1);
+    
+    assertEquals(instanceId, workflowOutput.getWorkflowId());
+  }
+
   private TestWorkflowPayload deserialize(String value) throws JsonProcessingException {
     return OBJECT_MAPPER.readValue(value, TestWorkflowPayload.class);
   }

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

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2025 The Dapr Authors
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+limitations under the License.
+*/
+package io.dapr.it.testcontainers.workflows;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class KeyStore {
+
+    private final Map<String, Boolean> keyStore = new HashMap<>();
+
+    private static KeyStore instance;
+
+    private KeyStore() {
+    }
+
+    public static KeyStore getInstance() {
+        if (instance == null) {
+            synchronized (KeyStore.class) {
+                if (instance == null) {
+                    instance = new KeyStore();
+                }
+            }
+        }
+        return instance;
+    }
+    
+
+    public void addKey(String key, Boolean value) {
+        keyStore.put(key, value);
+    }
+
+    public Boolean getKey(String key) {
+        return keyStore.get(key);
+    }
+
+    public void removeKey(String key) {
+        keyStore.remove(key);
+    }
+
+    public int size() {
+        return keyStore.size();
+    }
+
+}

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

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2025 The Dapr Authors
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package io.dapr.it.testcontainers.workflows;
+
+import io.dapr.workflows.WorkflowActivity;
+import io.dapr.workflows.WorkflowActivityContext;
+
+public class TaskExecutionIdActivity implements WorkflowActivity {
+
+  @Override
+  public Object run(WorkflowActivityContext ctx) {
+    TestWorkflowPayload workflowPayload = ctx.getInput(TestWorkflowPayload.class);
+    KeyStore keyStore = KeyStore.getInstance();
+    Boolean exists = keyStore.getKey(ctx.getTaskExecutionId());
+    if (!Boolean.TRUE.equals(exists)) {
+      keyStore.addKey(ctx.getTaskExecutionId(), true);
+      workflowPayload.getPayloads().add("Execution key not found");
+      throw new IllegalStateException("Task execution key not found");
+    }
+    workflowPayload.getPayloads().add("Execution key found");
+    return workflowPayload;
+  }
+
+}

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

@@ -0,0 +1,58 @@
+/*
+ * Copyright 2025 The Dapr Authors
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package io.dapr.it.testcontainers.workflows;
+
+import io.dapr.durabletask.Task;
+import io.dapr.workflows.Workflow;
+import io.dapr.workflows.WorkflowStub;
+import io.dapr.workflows.WorkflowTaskOptions;
+import io.dapr.workflows.WorkflowTaskRetryPolicy;
+
+import java.time.Duration;
+
+import org.slf4j.Logger;
+
+public class TestExecutionKeysWorkflow implements Workflow {
+
+  @Override
+  public WorkflowStub create() {
+    return ctx -> {
+      
+      Logger logger = ctx.getLogger();
+      String instanceId = ctx.getInstanceId();
+      logger.info("Starting Workflow: " + ctx.getName());
+      logger.info("Instance ID: " + instanceId);
+      logger.info("Current Orchestration Time: " + ctx.getCurrentInstant());
+
+      TestWorkflowPayload workflowPayload = ctx.getInput(TestWorkflowPayload.class);
+      workflowPayload.setWorkflowId(instanceId);
+
+      WorkflowTaskOptions options = new WorkflowTaskOptions(WorkflowTaskRetryPolicy.newBuilder()
+      .setMaxNumberOfAttempts(3)
+      .setFirstRetryInterval(Duration.ofSeconds(1))
+      .setMaxRetryInterval(Duration.ofSeconds(10))
+      .setBackoffCoefficient(2.0)  
+      .setRetryTimeout(Duration.ofSeconds(50))
+      .build());
+            
+      
+      Task<TestWorkflowPayload> t =   ctx.callActivity(TaskExecutionIdActivity.class.getName(), workflowPayload, options,TestWorkflowPayload.class);
+
+      TestWorkflowPayload payloadAfterExecution = t.await();
+   
+      ctx.complete(payloadAfterExecution);
+    };
+  }
+
+}

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

@@ -50,40 +50,41 @@ public WorkflowRuntimeBuilder workflowRuntimeBuilder(
       @Value("${dapr.http.endpoint}") String daprHttpEndpoint,
       @Value("${dapr.grpc.endpoint}") String daprGrpcEndpoint
   ){
-    Map<String, String> overrides = Map.of(
-        "dapr.http.endpoint", daprHttpEndpoint,
-        "dapr.grpc.endpoint", daprGrpcEndpoint
-    );
-
-    WorkflowRuntimeBuilder builder = new WorkflowRuntimeBuilder(new Properties(overrides));
+      Map<String, String> overrides = Map.of(
+              "dapr.http.endpoint", daprHttpEndpoint,
+              "dapr.grpc.endpoint", daprGrpcEndpoint
+      );
 
-    builder.registerWorkflow(TestWorkflow.class);
-    builder.registerWorkflow(TestNamedActivitiesWorkflow.class);
+      WorkflowRuntimeBuilder builder = new WorkflowRuntimeBuilder(new Properties(overrides));
 
-    builder.registerActivity(FirstActivity.class);
-    builder.registerActivity(SecondActivity.class);
-    builder.registerActivity("a",FirstActivity.class);
-    builder.registerActivity("b",FirstActivity.class);
-    builder.registerActivity("c", new SecondActivity());
-    builder.registerActivity("d", new WorkflowActivity() {
-      @Override
-      public Object run(WorkflowActivityContext ctx) {
-        TestWorkflowPayload workflowPayload = ctx.getInput(TestWorkflowPayload.class);
-        workflowPayload.getPayloads().add("Anonymous Activity");
-        return workflowPayload;
-      }
-    });
-    builder.registerActivity("e", new WorkflowActivity() {
-      @Override
-      public Object run(WorkflowActivityContext ctx) {
-        TestWorkflowPayload workflowPayload = ctx.getInput(TestWorkflowPayload.class);
-        workflowPayload.getPayloads().add("Anonymous Activity 2");
-        return workflowPayload;
-      }
-    });
+      builder.registerWorkflow(TestWorkflow.class);
+      builder.registerWorkflow(TestExecutionKeysWorkflow.class);
+      builder.registerWorkflow(TestNamedActivitiesWorkflow.class);
 
+      builder.registerActivity(FirstActivity.class);
+      builder.registerActivity(SecondActivity.class);
+      builder.registerActivity(TaskExecutionIdActivity.class);
 
+      builder.registerActivity("a", FirstActivity.class);
+      builder.registerActivity("b", FirstActivity.class);
+      builder.registerActivity("c", new SecondActivity());
+      builder.registerActivity("d", new WorkflowActivity() {
+          @Override
+          public Object run(WorkflowActivityContext ctx) {
+              TestWorkflowPayload workflowPayload = ctx.getInput(TestWorkflowPayload.class);
+              workflowPayload.getPayloads().add("Anonymous Activity");
+              return workflowPayload;
+          }
+      });
+      builder.registerActivity("e", new WorkflowActivity() {
+          @Override
+          public Object run(WorkflowActivityContext ctx) {
+              TestWorkflowPayload workflowPayload = ctx.getInput(TestWorkflowPayload.class);
+              workflowPayload.getPayloads().add("Anonymous Activity 2");
+              return workflowPayload;
+          }
+      });
 
-    return builder;
+      return builder;
   }
 }

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

@@ -17,6 +17,8 @@ public interface WorkflowActivityContext {
 
   String getName();
 
+  String getTaskExecutionId();
+
   <T> T getInput(Class<T> targetType);
 
 }

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

@@ -56,4 +56,9 @@ public String getName() {
   public <T> T getInput(Class<T> targetType) {
     return this.innerContext.getInput(targetType);
   }
+
+  @Override
+  public String getTaskExecutionId() {
+    return this.innerContext.getTaskExecutionId();
+  }
 }