Lack of Configurable Retry Logic for Activity Method

[rishi-dev89]

Currently, all exceptions thrown by the Activity method are treated as non-retryable, preventing the client from leveraging Dapr's built-in retry mechanism. Ideally, there should be an exception hierarchy where specific exceptions are classified as retryable.

https://github.com/dapr/durabletask-java/blob/main/client/src/main/java/io/dapr/durabletask/DurableTaskGrpcWorker.java#L168

[siri-varma]

/assign

[rishi-dev89]

Steps to reproduce in locally:
`
public class ActivityFailureDemoWorkflow implements Workflow {
@OverRide
public WorkflowStub create() {
return ctx -> {
ctx.getLogger().info("Starting Workflow: " + ctx.getName());
WorkflowTaskRetryPolicy retryPolicy = WorkflowTaskRetryPolicy.newBuilder()
.setMaxNumberOfAttempts(2)
.setFirstRetryInterval(Duration.ofSeconds(1))
.build();
String result;
WorkflowTaskOptions taskOptions = new WorkflowTaskOptions(retryPolicy);
try {
result = ctx.callActivity(ActivityThrowingException.class.getName(), "Tokyo", taskOptions, String.class).await() + ", ";
} catch (OrchestratorBlockedException ex) {
throw ex;
} catch (Throwable e) {
ctx.getLogger().error("Caught exception: " + e.getMessage(), e);
throw e;
}
ctx.getLogger().info("Workflow finished with result: " + result);
ctx.complete(result);
};
}
}

public class CustomException extends RuntimeException {
public CustomException(String message) {
super(message);
}
}

public class ActivityThrowingException implements WorkflowActivity {
@OverRide
public Object run(WorkflowActivityContext ctx) {
Logger logger = LoggerFactory.getLogger(ActivityThrowingException.class);
logger.info("Starting Activity: " + ctx.getName());
var message = ctx.getInput(String.class);
var newMessage = message.toUpperCase();
logger.info("Message Received from input: " + message);
logger.info("Sending message to output: " + newMessage);
throw new CustomException("This is a test exception");
}
}

public class DemoFailiureClient {
/**
* The main method to start the client.
*
* @param args Input arguments (unused).
* @throws InterruptedException If program has been interrupted.
*/
public static void main(String[] args) {
try (DaprWorkflowClient client = new DaprWorkflowClient()) {
String instanceId = client.scheduleNewWorkflow(ActivityFailureDemoWorkflow.class);
System.out.printf("Started a new chaining model workflow with instance ID: %s%n", instanceId);
WorkflowInstanceStatus workflowInstanceStatus = client.waitForInstanceCompletion(instanceId, null, true);

        String result = workflowInstanceStatus.readOutputAs(String.class);
        System.out.printf("workflow instance with ID: %s completed with result: %s%n", instanceId, result);


    } catch (TimeoutException | InterruptedException e) {
        System.out.println("Workflow execution timed out or was interrupted.");
        e.printStackTrace();
        throw new RuntimeException(e);
    }
}

}

public class DemoFailiureWorker {
/**

• The main method of this app.
• @param args The port the app will listen on.
• @throws Exception An Exception.
  */
  public static void main(String[] args) throws Exception {
  // Register the Workflow with the builder.
  WorkflowRuntimeBuilder builder = new WorkflowRuntimeBuilder().registerWorkflow(ActivityFailureDemoWorkflow.class);
  builder.registerActivity(ActivityThrowingException.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();
}

}
`

Result Output :

2025-05-12 19:34:18,579 {HH:mm:ss.SSS} [main] INFO i.d.w.runtime.WorkflowRuntimeBuilder - Registered Workflow: ActivityFailureDemoWorkflow
2025-05-12 19:34:18,586 {HH:mm:ss.SSS} [main] INFO i.d.w.runtime.WorkflowRuntimeBuilder - Registered Activity: ActivityThrowingException
2025-05-12 19:34:18,704 {HH:mm:ss.SSS} [main] INFO i.d.w.runtime.WorkflowRuntimeBuilder - List of registered workflows: [io.dapr.examples.workflows.chain.ActivityFailureDemoWorkflow]
2025-05-12 19:34:18,704 {HH:mm:ss.SSS} [main] INFO i.d.w.runtime.WorkflowRuntimeBuilder - List of registered activities: [io.dapr.examples.workflows.chain.ActivityThrowingException]
2025-05-12 19:34:18,704 {HH:mm:ss.SSS} [main] INFO i.d.w.runtime.WorkflowRuntimeBuilder - Successfully built dapr workflow runtime
Start workflow runtime
May 12, 2025 7:34:18 PM io.dapr.durabletask.DurableTaskGrpcWorker startAndBlock
INFO: Durable Task worker is connecting to sidecar at 127.0.0.1:50001.
2025-05-12 19:34:25,748 {HH:mm:ss.SSS} [pool-2-thread-1] INFO io.dapr.workflows.WorkflowContext - Starting Workflow: io.dapr.examples.workflows.chain.ActivityFailureDemoWorkflow
2025-05-12 19:34:25,797 {HH:mm:ss.SSS} [pool-2-thread-1] INFO i.d.e.w.c.ActivityThrowingException - Starting Activity: io.dapr.examples.workflows.chain.ActivityThrowingException
2025-05-12 19:34:25,799 {HH:mm:ss.SSS} [pool-2-thread-1] INFO i.d.e.w.c.ActivityThrowingException - Message Received from input: Tokyo
2025-05-12 19:34:25,799 {HH:mm:ss.SSS} [pool-2-thread-1] INFO i.d.e.w.c.ActivityThrowingException - Sending message to output: TOKYO
2025-05-12 19:34:25,831 {HH:mm:ss.SSS} [pool-2-thread-1] ERROR io.dapr.workflows.WorkflowContext - Caught exception: Task 'io.dapr.examples.workflows.chain.ActivityThrowingException' (#0) failed with an unhandled exception: This is a test exception
io.dapr.durabletask.TaskFailedException: Task 'io.dapr.examples.workflows.chain.ActivityThrowingException' (#0) failed with an unhandled exception: This is a test exception
at io.dapr.durabletask.TaskOrchestrationExecutor$ContextImplTask.handleTaskFailed(TaskOrchestrationExecutor.java:539)
at io.dapr.durabletask.TaskOrchestrationExecutor$ContextImplTask.processEvent(TaskOrchestrationExecutor.java:865)
at io.dapr.durabletask.TaskOrchestrationExecutor$ContextImplTask.access$1300(TaskOrchestrationExecutor.java:74)
at io.dapr.durabletask.TaskOrchestrationExecutor$ContextImplTask$OrchestrationHistoryIterator.moveNext(TaskOrchestrationExecutor.java:958)
at io.dapr.durabletask.TaskOrchestrationExecutor$ContextImplTask.processNextEvent(TaskOrchestrationExecutor.java:818)
at io.dapr.durabletask.TaskOrchestrationExecutor$ContextImplTask.access$000(TaskOrchestrationExecutor.java:74)
at io.dapr.durabletask.TaskOrchestrationExecutor$ContextImplTask$CompletableTask.processNextEvent(TaskOrchestrationExecutor.java:1308)
at io.dapr.durabletask.TaskOrchestrationExecutor$ContextImplTask$CompletableTask.await(TaskOrchestrationExecutor.java:1296)
at io.dapr.durabletask.TaskOrchestrationExecutor$ContextImplTask$RetriableTask.await(TaskOrchestrationExecutor.java:1155)
at io.dapr.examples.workflows.chain.ActivityFailureDemoWorkflow.lambda$create$0(ActivityFailureDemoWorkflow.java:40)
at io.dapr.workflows.Workflow.run(Workflow.java:37)
at io.dapr.workflows.runtime.WorkflowClassWrapper.lambda$create$0(WorkflowClassWrapper.java:60)
at io.dapr.durabletask.TaskOrchestrationExecutor$ContextImplTask.processEvent(TaskOrchestrationExecutor.java:849)
at io.dapr.durabletask.TaskOrchestrationExecutor$ContextImplTask.access$1300(TaskOrchestrationExecutor.java:74)
at io.dapr.durabletask.TaskOrchestrationExecutor$ContextImplTask$OrchestrationHistoryIterator.moveNext(TaskOrchestrationExecutor.java:958)
at io.dapr.durabletask.TaskOrchestrationExecutor$ContextImplTask.processNextEvent(TaskOrchestrationExecutor.java:818)
at io.dapr.durabletask.TaskOrchestrationExecutor$ContextImplTask.access$000(TaskOrchestrationExecutor.java:74)
at io.dapr.durabletask.TaskOrchestrationExecutor.execute(TaskOrchestrationExecutor.java:52)
at io.dapr.durabletask.DurableTaskGrpcWorker.lambda$startAndBlock$0(DurableTaskGrpcWorker.java:133)
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:572)
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:317)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
at java.base/java.lang.Thread.run(Thread.java:1583)
May 12, 2025 7:34:25 PM io.dapr.durabletask.TaskOrchestrationExecutor execute
WARNING: The orchestrator failed with an unhandled exception: io.dapr.durabletask.TaskFailedException: Task 'io.dapr.examples.workflows.chain.ActivityThrowingException' (#0) failed with an unhandled exception: This is a test exception

[siri-varma]

@rishi-dev89 Not setting "setRetryTimeout" is causing an NPE because of which retries are not working. For now, please supply the "setRetryTimeout" and that should unblock you folks.

I have also raised a PR to fix this holistically.

Thank you for raising this

[siri-varma]

Durable task PR: dapr/durabletask-java#17