Dapr Workflows With PubSub

1. Introduction

Our previous article explored how to build a loosely coupled PubSub messaging system using Spring Boot and Dapr with a ride-hailing application. While this approach works well for simple message routing, real-world scenarios often require more sophisticated orchestration.

In this tutorial, we’ll extend our scenario by introducing Dapr Workflows. We’ll learn how to orchestrate complex, long-running processes that react to REST endpoints and PubSub events while maintaining durability. We’ll also demonstrate how to test these workflows using Testcontainers, ensuring our system works as expected without external infrastructure.

2. Understanding Durable Execution

Traditional approaches to this problem often involve:

• Manual state management in databases
• Complex retry logic scattered across services
• Custom error handling for each operation
• Difficulty tracking workflow progress
• Loss of context when services restart mid-process

2.1. Solving the Problem With Dapr Workflows

Dapr Workflows solves these challenges through durable execution, which is a combination of solutions:

• Automatic State Persistence: The workflow engine persists state after each step. If a process crashes or restarts, it automatically resumes from where it left off without losing progress. This happens transparently as the runtime handles the complexity.
• Workflow Replay: When a workflow resumes after a failure, Dapr replays the workflow code from the beginning. However, instead of re-executing completed activities, it uses the persisted results. This means our workflow orchestration logic runs multiple times, which is why it must be deterministic.
• Built-in Resilience: Rather than implementing custom retry logic and error handling for each operation, we define retry policies once. Dapr automatically applies these policies, manages timeouts, and handles transient failures.
• Simple Orchestration: Complex processes become straightforward code. We can write sequential logic with await() calls, use conditional branching, and loop through steps.

3. Workflows as Code

Dapr Workflows lets us define orchestration logic using familiar programming constructs rather than XML or JSON definitions. The Spring Boot integration makes this seamless by managing workflows and activities as Spring beans.

3.1. The Workflow Abstraction

A workflow represents the sequence of steps we need to execute. In Dapr, we implement the Workflow interface and define our logic using the create() method. The workflow context provides methods for calling activities, waiting for external events, and managing workflow state. We haven’t created activities yet, so let’s start with our task options:

@Component
public class RideProcessingWorkflow implements Workflow {
    
    @Override
    public WorkflowStub create() {
        return context -> {
            WorkflowTaskOptions options = taskOptions();
            // ...
        };
    }
}

3.2. The WorkflowActivity Abstraction

Activities encapsulate individual work units within a workflow, representing tasks that interact with external systems, databases, or services. Our first activity is a simple validation of the driver’s ID by getting the RideWorkflowRequest from our context:

@Component
public class ValidateDriverActivity implements WorkflowActivity {
    
    @Autowired
    private RestTemplate restTemplate;
    
    @Override
    public Object run(WorkflowActivityContext context) {
        RideWorkflowRequest request = ctx.getInput(RideWorkflowRequest.class);
        logger.info("Validating driver: {}", request.getDriverId());

        if (request.getDriverId() != null && !request.getDriverId().isEmpty()) {
            logger.info("Driver {} validated successfully", request.getDriverId());
            return true;
        }

        throw new IllegalArgumentException("Invalid driver ID");
    }
}

Since activities are components, we can leverage dependency injection to access other beans like repositories or any service we need. Activities can fail, and the workflow runtime handles their execution and state management. Here, we throw an exception if the validation fails.

3.3. Enabling Workflow Auto-Discovery

The @EnableDaprWorkflows annotation simplifies workflow registration by automatically discovering all workflows and activities on the classpath. Let’s use it on our Spring Boot app class:

@EnableDaprWorkflows
@SpringBootApplication
public class DaprWorkflowApp {

    public static void main(String[] args) {
        SpringApplication.run(DaprWorkflowApp.class, args);
    }
}

With this annotation in place, Spring Boot automatically:

• Scans for classes implementing Workflow and WorkflowActivity
• Registers them as Spring beans
• Makes them available to the Dapr workflow runtime
• Enables dependency injection throughout the workflow infrastructure

This eliminates boilerplate configuration and follows Spring Boot’s convention-over-configuration philosophy.

4. Extending the Ride-Hailing PubSub Example

Let’s enhance our ride-hailing application to include a workflow orchestrating the entire ride lifecycle.

4.1. The Scenario

When a driver accepts a ride request, we want to:

1. Validate the driver’s credentials
2. Calculate and process the estimated fare
3. Notify the passenger that a driver is en route
4. Update the ride status to “in progress”

This workflow starts when we receive a “driver accepted” event from our PubSub system. Each step needs proper error handling, and the entire process must be durable. If our service restarts, the workflow should resume seamlessly.

4.2. Defining the Domain Models

We’ll wrap our RideRequest model in a RideWorkflowRequest object to include workflow-specific information:

public class RideWorkflowRequest {
    private RideRequest rideRequest;
    private String rideId;
    private String driverId;
    private String workflowInstanceId;
    
    // constructors, getters, and setters
}

Finally, we’ll create a simple model to define progress when our workflow completes:

public record RideWorkflowStatus(String rideId, String status, String message) {}

4.3. Starting Workflows via REST

First, let’s create a REST endpoint to start a workflow. Our controller only needs the DaprWorkflowClient, and our endpoint receives the input data we need for context:

@RestController
@RequestMapping("/workflow")
public class RideWorkflowController {

    @Autowired
    DaprWorkflowClient workflowClient;

    @PostMapping("/start-ride")
    public RideWorkflowRequest startRideWorkflow(
      @RequestBody RideWorkflowRequest request) {
        // ...
        return request;
    }
}

The DaprWorkflowClient allows us to schedule new workflow instances and query their status. Each workflow gets a unique instance ID that we can use to track its progress. Let’s start the workflow by calling scheduleNewWorkflow(), passing our workflow class and context data:

String instanceId = workflowClient.scheduleNewWorkflow(
  RideProcessingWorkflow.class, request); 
request.setWorkflowInstanceId(instanceId);

4.4. Raising Workflow Events

We can trigger an event from anywhere if we have the workflow instance ID. Let’s create an endpoint used by passengers that’ll trigger a confirmation along with a String payload:

@PostMapping("/confirm/{instanceId}")
public void confirmRide(
  @PathVariable("instanceId") String instanceId, @RequestBody String confirmation) {
    workflowClient.raiseEvent(instanceId, "passenger-confirmation", confirmation);
}

Raising events is usually used with waitForExternalEvent() from the WorkflowContext for orchestration.

4.5. Triggering Workflows From PubSub Events

Creating a subscriber that listens for ride acceptance events looks very similar. The only difference is that we use the @Topic annotation and wrap our input in a CloudEvent:

@PostMapping("/driver-accepted")
@Topic(pubsubName = "ride-hailing", name = "driver-acceptance")
public void onDriverAcceptance(@RequestBody CloudEvent<RideWorkflowRequest> event) {
    RideWorkflowRequest request = event.getData();
    workflowClient.scheduleNewWorkflow(RideProcessingWorkflow.class, request);
}

5. Implementing More Workflow Activities

Let’s continue creating activities for each step in our workflow.

5.1. Calculating Fares

Next, we’ll create an activity to calculate the fare:

@Component
public class CalculateFareActivity implements WorkflowActivity {

    @Override
    public Object run(WorkflowActivityContext context) {
        double baseFare = 5.0;
        double perMileFare = 2.5;
        double estimatedMiles = 10.0;
        
        return baseFare + (perMileFare * estimatedMiles);
    }
}

5.2. Notifying Passengers

Finally, let’s create an activity to notify the passenger. We’ll use a record to store all the data we need:

public record NotificationInput(RideWorkflowRequest request, double fare) {}

In a real application, we’d send a notification via email, push, or anything else. To focus on the framework, we’ll only log the message:

@Component
public class NotifyPassengerActivity implements WorkflowActivity {

    @Override
    public Object run(WorkflowActivityContext context) {
        NotificationInput input = context.getInput(NotificationInput.class);

        String message = String.format(
          "Driver %s is on the way to %s. Estimated fare: $%.2f", 
          input.request().getDriverId(),
          input.request().getRideRequest().getLocation(),
          input.fare());

        context.getLogger().info("Notification sent: {}", message);
        return message;
    }
}

6. Orchestrating the Workflow

Now that we have everything we need to complete our workflow, let’s return to our RideProcessingWorkflow class to implement the orchestration for our activities.

6.1. Validating the Driver

We’ll log the current step and check the boolean returned by the ValidateDriverActivity:

context.getLogger().info("Step 1: Validating driver {}", request.getDriverId());
boolean isValid = context.callActivity(
  ValidateDriverActivity.class.getName(), request, options, boolean.class)
    .await();

If the activity returns false, we’ll complete the workflow with a “FAILED” status:

if (!isValid) {
    context.complete(new RideWorkflowStatus(
      request.getRideId(), "FAILED", "Driver validation failed"));
    return;
}

6.2. Calculating the Fare and Notifying the Passenger

First, we obtain the value for the fare by calling our CalculateFareActivity:

context.getLogger().info("Step 2: Calculating fare");
double fare = context.callActivity(
  CalculateFareActivity.class.getName(), request, options, double.class)
    .await();

Then use it to build our NotificationInput and call its activity:

context.getLogger().info("Step 3: Notifying passenger");
NotificationInput notificationInput = new NotificationInput(request, fare);
String notification = context.callActivity(
  NotifyPassengerActivity.class.getName(), notificationInput, options, String.class)
    .await();

6.3. Waiting for Passenger Confirmation

To block until an event is triggered, we call waitForExternalEvent() on our context and optionally a timeout value. If we want to access the event’s payload, we also need to specify its type:

context.getLogger().info("Step 4: Waiting for passenger confirmation");
String confirmation = context.waitForExternalEvent(
  "passenger-confirmation", Duration.ofMinutes(5), String.class)
    .await();

We expect the confirmation payload to be precisely “confirmed” for this example. Otherwise, we complete with a “CANCELLED” status:

if (!"confirmed".equalsIgnoreCase(confirmation)) {
    context.complete(new RideWorkflowStatus(
      request.getRideId(), 
      "CANCELLED", 
      "Passenger did not confirm the ride within the timeout period"));
    return;
}

6.4. Completing the Workflow

If everything goes OK, we’ll finally complete the workflow with a “COMPLETED” status:

String message = String.format(
  "Ride confirmed and processed successfully. Fare: $%.2f. %s", fare, notification);
RideWorkflowStatus status = new RideWorkflowStatus(
  request.getRideId(), "COMPLETED", message);

context.getLogger().info("Workflow completed: {}", message);
context.complete(status);

7. Testing Workflows With Testcontainers

Testing workflows is crucial to ensure our orchestration logic works correctly. The dapr-spring-boot-starter-test module integrates seamlessly with Testcontainers, allowing us to test the complete system without external CLI tools.

7.1. Testing Our Happy Path

So let’s test our entire workflow, since the first REST call to “start-ride”:

@Test
void whenWorkflowStartedViaRest_thenAllActivitiesExecute() {
    RideRequest rideRequest = new RideRequest(
      "passenger-1", "Downtown", "Airport");
    RideWorkflowRequest workflowRequest = new RideWorkflowRequest(
      "ride-123", rideRequest, "driver-456", null);

    RideWorkflowRequest response = given().contentType(ContentType.JSON)
      .body(workflowRequest)
      .when()
      .post("/workflow/start-ride")
      .then()
      .statusCode(200)
      .extract()
      .as(RideWorkflowRequest.class);

    String instanceId = response.getWorkflowInstanceId();
    assertNotNull(instanceId);

    // ...
}

After the response, we store the workflow instance ID and wait until the workflow is running by checking workflowClient.getInstanceState() with awaitility:

await().atMost(Duration.ofSeconds(10))
  .pollInterval(Duration.ofMillis(200))
  .until(() -> {
    WorkflowInstanceStatus status = workflowClient.getInstanceState(instanceId, false);
    return status != null && status.getRuntimeStatus() == WorkflowRuntimeStatus.RUNNING;
  });

Then, we raise the passenger confirmation event so the workflow can resume:

given().contentType(ContentType.TEXT)
  .body("confirmed")
  .when()
  .post("/workflow/confirm/" + instanceId)
  .then()
  .statusCode(200);

In the end, we wait for the workflow to complete by rechecking the instance state:

await().atMost(Duration.ofSeconds(15))
  .pollInterval(Duration.ofMillis(500))
  .until(() -> {
      WorkflowInstanceStatus status = workflowClient.getInstanceState(instanceId, false);
      return status != null && status.getRuntimeStatus() == WorkflowRuntimeStatus.COMPLETED;
  });

Finally, we verify that the workflow completed successfully by checking the instance state and status:

WorkflowInstanceStatus finalStatus = workflowClient.getInstanceState(instanceId, true);
assertEquals(WorkflowRuntimeStatus.COMPLETED, finalStatus.getRuntimeStatus());

7.2. Building Our Retry Policy

Let’s see how to leverage Dapr’s durable execution model and built-in support for timeouts, retries, and compensation. We’ll define the retry policy used for every step in our workflow by implementing the taskOptions() method we defined in RideProcessingWorkflow. We’ll define a max retries, backoff timeout, retry interval, and max timeout by instantiating a WorkflowTaskRetryPolicy and returning it in a reusable WorkflowTaskOptions:

private WorkflowTaskOptions taskOptions() {
    int maxRetries = 3;
    Duration backoffTimeout = Duration.ofSeconds(1);
    double backoffCoefficient = 1.5;
    Duration maxRetryInterval = Duration.ofSeconds(5);
    Duration maxTimeout = Duration.ofSeconds(10);

    WorkflowTaskRetryPolicy retryPolicy = new WorkflowTaskRetryPolicy(
      maxRetries, backoffTimeout, backoffCoefficient, maxRetryInterval, maxTimeout);
    return new WorkflowTaskOptions(retryPolicy);
}

@Test
void whenActivityFails_thenRetryPolicyApplies() {
    RideWorkflowRequest invalidRequest = new RideWorkflowRequest(
      "ride-789", new RideRequest("passenger-3", "Park", "Beach"), "", null);

    String instanceId = workflowClient.scheduleNewWorkflow(
      RideProcessingWorkflow.class, invalidRequest);

    await().atMost(Duration.ofSeconds(20))
      .pollInterval(Duration.ofMillis(500))
      .until(() -> {
        WorkflowInstanceStatus status = workflowClient.getInstanceState(instanceId, false);
        return status != null && status.getRuntimeStatus() == WorkflowRuntimeStatus.FAILED;
    });

    WorkflowInstanceStatus finalStatus = workflowClient.getInstanceState(instanceId, true);
    assertEquals(WorkflowRuntimeStatus.FAILED, finalStatus.getRuntimeStatus());
}

8. Running With Catalyst on the Diagrid CLI

After local development, setting up and maintaining infrastructure ourselves involves significant operational complexity. Diagrid Catalyst eliminates this burden by providing a fully managed Dapr environment.

8.1. Setting up and Running Our App

Once we register a free account, we can install the Diagrid CLI and run our application with:

diagrid dev run \
  --project spring-boot \
  --app-id dapr-workflows \
  --app-port 60603 \
  -- mvn spring-boot:run

This command:

• Creates a new project/app
• Connects our local application to Catalyst’s managed Dapr services
• Uses cloud-based workflow orchestration and state management with production-grade infrastructure

9. Conclusion

In this article, we’ve extended our Dapr PubSub example to include durable workflow orchestration. By combining Dapr PubSub with Dapr Workflows, we created a system that handles event-driven messaging and complex multi-step processes.

As always, the source code is available over on GitHub.