Skip to content

Add taskName and input to retry context #1423

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 4 commits into
base: master
Choose a base branch
from
+353 −13
Open

Add taskName and input to retry context #1423

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
83a58d4
Add taskName and input to retry handler
TheForbiddenAi Jun 13, 2025
fcb5086
Add retry handler example
TheForbiddenAi Jun 16, 2025
23039f1
Update example README file
TheForbiddenAi Jun 16, 2025
30a41cd
Merge branch 'master' into add_info_to_handler
siri-varma Jun 16, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 48 additions & 0 deletions examples/src/main/java/io/dapr/examples/workflows/retryhandler/DemoRetryHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,48 @@
/*
* 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.examples.workflows.retryhandler;

import io.dapr.workflows.WorkflowContext;
import io.dapr.workflows.WorkflowTaskRetryContext;
import io.dapr.workflows.WorkflowTaskRetryHandler;
import org.slf4j.Logger;

import java.time.Duration;
import java.time.Instant;
import java.util.concurrent.TimeUnit;

public class DemoRetryHandler implements WorkflowTaskRetryHandler {

@Override
public boolean handle(WorkflowTaskRetryContext retryContext) {
WorkflowContext workflowContext = retryContext.getWorkflowContext();
Logger logger = retryContext.getWorkflowContext().getLogger();
Object input = retryContext.getInput();
String taskName = retryContext.getTaskName();

if(taskName.equalsIgnoreCase(FailureActivity.class.getName())) {
Copy link
Contributor

@siri-varma siri-varma Jun 16, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Interesting, so we can have multiple if blocks

if (taskName.equals..(ChildWorkf1.class)) {
// do something
}

if (taskName.equals..(ChildWorkf2.class)) {
// do something
}

if (taskName.equals..(ChildActivity1.class)) {
// do something
}

Wouldn't it be more maintainable to colocate the workflow or activity with its corresponding retry logic? While I understand the code duplication part, centralizing all retry handling in a single utility class does not seem optimal

@cicoyle , @artur-ciocanu , @salaboy I would like to know your thoughts too.

Copy link
Contributor Author

@TheForbiddenAi TheForbiddenAi Jun 16, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So in the case of multiple activites, I would recommend breaking them up into separate retry handlers that call each other (See chain of responsibility design pattern). The main benefit of chaining handlers is that you can reuse logic from prior handlers to reduce code duplication. For example

class HandlerOne implements WorkflowTaskRetryHandler {

  private WorkflowTaskRetryHandler nextHandler;
  
  public boolean handle(WorkflowTaskRetryContextcontext) {
    // handler logic here
    if(nextHandler != null) {
      return nextHandler.handle(context);
    }
    return true;
  }

  void setNextHandler(WorkflowTaskRetryHandler handler) {
     this.nextHandler = handler;
  }
}

In a workflow for instance, it would look like this (HandlerTwo follows same layout as HandlerOne above)

public class TestWorkflow implements Workflow {

  @Override
  public WorkflowStub create() {
    return context -> {
      Logger logger = context.getLogger();
      logger.info("Starting RetryWorkflow: {}", context.getName());

      HandlerOne handlerOne = new HandlerOne();
      HandlerTwo handlerOne = new HandlerTwo();

      handlerOne.setNextHandler(handlerTwo);
      WorkflowTaskOptions taskOptions = new WorkflowTaskOptions(handlerOne);

      logger.info("RetryWorkflow is calling Activity: {}", FailureActivity.class.getName());
      Instant currentTime = context.getCurrentInstant();
      Instant result = context.callActivity(FailureActivity.class.getName(), currentTime, taskOptions, Instant.class).await();

      logger.info("RetryWorkflow finished with: {}",  result);
      context.complete(result);
    };
  }

}

Ideally, you would define an interface or abstract class that defines any common methods/logic (i.e. setNextHandler) when using this pattern, but for the sake of simplicity, I left it out in the example above

Copy link
Contributor Author

@TheForbiddenAi TheForbiddenAi Jun 16, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Additionally, I'm curious about scenarios where retries are driven solely by the names of workflow/activities. Could you share an example where that pattern would be more effective?

Perhaps, you want to have variable backoff times depending on what each activity is doing. Additionally, if you know the name of the activity, you know the type of its input. Also, having the name is useful when it comes to tracking what task is being retried in the logs, especially if you are running multiple activities concurrently.

Another possible situation is: let's say you have Task A. This task makes a request to a service under the assumption that this service has access to other data. If this service doesn't have this data, then Task A fails. To recover from this failure, you may want to spin off a new activity, we'll call this Task B, to do some action that creates that data.

In this hypothetical situation, you would only want Task B to be called if Task A fails to prevent duplicating data. Once Task B completes, we could then return to our retry handler to retry Task A automatically.

Copy link
Contributor

@siri-varma siri-varma Jun 17, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@TheForbiddenAi I was revisiting this comment earlier this morning and had a couple of thoughts:

Regarding logs: if you include a logger statement inside each activity, it should help you identify which activity is currently running, right?

As for the second scenario, one approach could be to structure the workflow so that it starts with Activity A. If Activity A fails due to missing data, the workflow can then trigger Activity B. Once B completes successfully, you can start Activity A within the same workflow.

Also, the way I see retries, they are for transient failures but in your case Activity A fails with permanent failure

Let me know your thoughts.

Copy link
Contributor Author

@TheForbiddenAi TheForbiddenAi Jun 17, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So for logs, if you have multiple workflows running or multiple activities running asynchronously. It would be very difficult to differentiate logs in the retry handler from one another.

As for second scenario, yes that is a valid approach. However it would require a try catch and then manually calling the activity again, meaning if you had multiple of these scenarios for the same activity, you would have to chain try catches which would not be great from a readability and maintainability standpoint

logger.info("FailureActivity Input: {}", input);
Instant timestampInput = (Instant) input;
// Add a second to ensure, it is 100% passed the time to success
Instant timeToSuccess = timestampInput.plusSeconds(FailureActivity.TIME_TO_SUCCESS + 1);
long timeToWait = timestampInput.until(timeToSuccess, TimeUnit.SECONDS.toChronoUnit());

logger.info("Waiting {} seconds before retrying.", timeToWait);
workflowContext.createTimer(Duration.ofSeconds(timeToWait)).await();
logger.info("Send request to FailureActivity");
}

return true;
}
}
47 changes: 47 additions & 0 deletions examples/src/main/java/io/dapr/examples/workflows/retryhandler/DemoRetryHandlerClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,47 @@
/*
* 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.examples.workflows.retryhandler;

import io.dapr.workflows.client.DaprWorkflowClient;
import io.dapr.workflows.client.WorkflowInstanceStatus;

import java.time.Duration;
import java.time.Instant;
import java.util.concurrent.TimeoutException;

public class DemoRetryHandlerClient {
/**
* 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(DemoRetryWorkflow.class);
System.out.printf("Started a new external-event workflow with instance ID: %s%n", instanceId);

// Block until the orchestration completes. Then print the final status, which includes the output.
WorkflowInstanceStatus workflowInstanceStatus = client.waitForInstanceCompletion(
instanceId,
Duration.ofSeconds(30),
true);

System.out.printf("workflow instance with ID: %s completed with result: %s%n", instanceId,
workflowInstanceStatus.readOutputAs(Instant.class));
} catch (TimeoutException | InterruptedException e) {
throw new RuntimeException(e);
}
}
}
36 changes: 36 additions & 0 deletions examples/src/main/java/io/dapr/examples/workflows/retryhandler/DemoRetryHandlerWorker.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,36 @@
/*
* 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.examples.workflows.retryhandler;

import io.dapr.workflows.runtime.WorkflowRuntime;
import io.dapr.workflows.runtime.WorkflowRuntimeBuilder;

public class DemoRetryHandlerWorker {
/**
* 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(DemoRetryWorkflow.class);
builder.registerActivity(FailureActivity.class);

// Build and then start the workflow runtime pulling and executing tasks
WorkflowRuntime runtime = builder.build();
System.out.println("Start workflow runtime");
runtime.start();
}
}
43 changes: 43 additions & 0 deletions examples/src/main/java/io/dapr/examples/workflows/retryhandler/DemoRetryWorkflow.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,43 @@
/*
* 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.examples.workflows.retryhandler;

import io.dapr.workflows.Workflow;
import io.dapr.workflows.WorkflowStub;
import io.dapr.workflows.WorkflowTaskOptions;
import io.dapr.workflows.WorkflowTaskRetryHandler;
import org.slf4j.Logger;

import java.time.Instant;

public class DemoRetryWorkflow implements Workflow {

@Override
public WorkflowStub create() {
return context -> {
Logger logger = context.getLogger();
logger.info("Starting RetryWorkflow: {}", context.getName());

WorkflowTaskRetryHandler retryHandler = new DemoRetryHandler();
WorkflowTaskOptions taskOptions = new WorkflowTaskOptions(retryHandler);

logger.info("RetryWorkflow is calling Activity: {}", FailureActivity.class.getName());
Instant currentTime = context.getCurrentInstant();
Instant result = context.callActivity(FailureActivity.class.getName(), currentTime, taskOptions, Instant.class).await();

logger.info("RetryWorkflow finished with: {}", result);
context.complete(result);
};
}
}
44 changes: 44 additions & 0 deletions examples/src/main/java/io/dapr/examples/workflows/retryhandler/FailureActivity.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,44 @@
/*
* 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.examples.workflows.retryhandler;

import io.dapr.workflows.WorkflowActivity;
import io.dapr.workflows.WorkflowActivityContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Instant;

public class FailureActivity implements WorkflowActivity {

private static final Logger LOGGER = LoggerFactory.getLogger(FailureActivity.class);
public static final long TIME_TO_SUCCESS = 10;

@Override
public Object run(WorkflowActivityContext ctx) {
LOGGER.info("Starting Activity: {}", ctx.getName());

Instant timestamp = ctx.getInput(Instant.class);

LOGGER.info("Input timestamp: {}", timestamp);
if(timestamp.plusSeconds(TIME_TO_SUCCESS).isBefore(Instant.now())) {
LOGGER.info("Completing Activity: {}", ctx.getName());
return Instant.now();
}

LOGGER.info("Throwing exception for Activity: {}", ctx.getName());

throw new RuntimeException("Failure!");
}
}
38 changes: 31 additions & 7 deletions sdk-workflows/src/main/java/io/dapr/workflows/WorkflowTaskRetryContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,16 +14,17 @@
package io.dapr.workflows;

import io.dapr.workflows.client.WorkflowFailureDetails;
import io.dapr.workflows.runtime.DefaultWorkflowContext;

import java.time.Duration;

public class WorkflowTaskRetryContext {

private final DefaultWorkflowContext workflowContext;
private final WorkflowContext workflowContext;
private final int lastAttemptNumber;
private final WorkflowFailureDetails lastFailure;
private final Duration totalRetryTime;
private final String taskName;
private final Object input;

/**
* Constructor for WorkflowTaskRetryContext.
Expand All @@ -32,28 +33,34 @@
* @param lastAttemptNumber The number of the previous attempt
* @param lastFailure The failure details from the most recent failure
* @param totalRetryTime The amount of time spent retrying
* @param taskName The name of the task
* @param input The input of the task
*/
public WorkflowTaskRetryContext(
DefaultWorkflowContext workflowContext,
WorkflowContext workflowContext,
int lastAttemptNumber,
WorkflowFailureDetails lastFailure,
Duration totalRetryTime) {
Duration totalRetryTime,
String taskName,
Object input) {
this.workflowContext = workflowContext;
this.lastAttemptNumber = lastAttemptNumber;
this.lastFailure = lastFailure;
this.totalRetryTime = totalRetryTime;
this.taskName = taskName;
this.input = input;
}

/**
* Gets the context of the current workflow.
*
* <p>The workflow context can be used in retry handlers to schedule timers (via the
* {@link DefaultWorkflowContext#createTimer} methods) for implementing delays between retries. It can also be
* used to implement time-based retry logic by using the {@link DefaultWorkflowContext#getCurrentInstant} method.
* {@link WorkflowContext#createTimer} methods) for implementing delays between retries. It can also be
* used to implement time-based retry logic by using the {@link WorkflowContext#getCurrentInstant} method.
*
* @return the context of the parent workflow
*/
public DefaultWorkflowContext getWorkflowContext() {
public WorkflowContext getWorkflowContext() {
return this.workflowContext;
}

Expand Down Expand Up @@ -85,4 +92,21 @@
return this.totalRetryTime;
}

/**
* Gets the name of the task.
*
* @return the name of the task
*/
public String getTaskName() {
return taskName;

Check warning on line 101 in sdk-workflows/src/main/java/io/dapr/workflows/WorkflowTaskRetryContext.java

View check run for this annotation

Codecov / codecov/patch

sdk-workflows/src/main/java/io/dapr/workflows/WorkflowTaskRetryContext.java#L101 

Added line #L101 was not covered by tests
}

/**
* Gets the input of the task.
*
* @return the task's input
*/
public Object getInput() {
return input;

Check warning on line 110 in sdk-workflows/src/main/java/io/dapr/workflows/WorkflowTaskRetryContext.java

View check run for this annotation

Codecov / codecov/patch

sdk-workflows/src/main/java/io/dapr/workflows/WorkflowTaskRetryContext.java#L110 

Added line #L110 was not covered by tests
}
}
18 changes: 12 additions & 6 deletions sdk-workflows/src/main/java/io/dapr/workflows/runtime/DefaultWorkflowContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -164,7 +164,7 @@ public boolean isReplaying() {
* {@inheritDoc}
*/
public <V> Task<V> callActivity(String name, Object input, WorkflowTaskOptions options, Class<V> returnType) {
TaskOptions taskOptions = toTaskOptions(options);
TaskOptions taskOptions = toTaskOptions(options, name, input);

return this.innerContext.callActivity(name, input, taskOptions, returnType);
}
Expand Down Expand Up @@ -208,7 +208,7 @@ public <T> T getInput(Class<T> targetType) {
@Override
public <V> Task<V> callChildWorkflow(String name, @Nullable Object input, @Nullable String instanceID,
@Nullable WorkflowTaskOptions options, Class<V> returnType) {
TaskOptions taskOptions = toTaskOptions(options);
TaskOptions taskOptions = toTaskOptions(options, name, input);

return this.innerContext.callSubOrchestrator(name, input, instanceID, taskOptions, returnType);
}
Expand Down Expand Up @@ -237,13 +237,13 @@ public UUID newUuid() {
return this.innerContext.newUUID();
}

private TaskOptions toTaskOptions(WorkflowTaskOptions options) {
private TaskOptions toTaskOptions(WorkflowTaskOptions options, String taskName, Object input) {
if (options == null) {
return null;
}

RetryPolicy retryPolicy = toRetryPolicy(options.getRetryPolicy());
RetryHandler retryHandler = toRetryHandler(options.getRetryHandler());
RetryHandler retryHandler = toRetryHandler(options.getRetryHandler(), taskName, input);

return new TaskOptions(retryPolicy, retryHandler);
}
Expand Down Expand Up @@ -276,9 +276,13 @@ private RetryPolicy toRetryPolicy(WorkflowTaskRetryPolicy workflowTaskRetryPolic
* Converts a {@link WorkflowTaskRetryHandler} to a {@link RetryHandler}.
*
* @param workflowTaskRetryHandler The {@link WorkflowTaskRetryHandler} being converted
* @param taskName The name of the task
* @param input The input object passed to the task
* @return A {@link RetryHandler}
*/
private RetryHandler toRetryHandler(WorkflowTaskRetryHandler workflowTaskRetryHandler) {
private RetryHandler toRetryHandler(WorkflowTaskRetryHandler workflowTaskRetryHandler,
String taskName,
Object input) {
if (workflowTaskRetryHandler == null) {
return null;
}
Expand All @@ -288,7 +292,9 @@ private RetryHandler toRetryHandler(WorkflowTaskRetryHandler workflowTaskRetryHa
this,
retryContext.getLastAttemptNumber(),
new DefaultWorkflowFailureDetails(retryContext.getLastFailure()),
retryContext.getTotalRetryTime()
retryContext.getTotalRetryTime(),
taskName,
input
);

return workflowTaskRetryHandler.handle(workflowRetryContext);
Expand Down
Loading