Batch Process

Perform a set of state transitions as a batch process.

Problem

In a bike rental domain a shipment of bikes has been delivered, with the bikes requiring registration in the rental service. The bikes need to be processed as one batch after which subsequent operations can commence.

There is thus a requirement to track the state transition - the "bike registered" event - for each bike and determine when all events have been successfully handled. Tracking should be amenable to both sequential and parallel processing.

Solution

The event model in Figure 1 shows a batch process for bike registration:

  • Store a unique batch identifier along with tracking status.

  • Associate the batch identifier with every RegisterBike command that is sent.

  • In the BatchStatus read model increment the noProcessed variable upon receiving a BikeRegistered event. If the number of handled events equal the expected number of state transitions the condition to continue with other operations is fulfilled.

figure 1

Figure 1 - Modeling a batch process

The batch identifier must be carried over from RegisterBike to BikeRegistered; if it is not possible to explicitly add the identifier as a field in these messages, the underlying messaging mechanism may support it via metadata (as is supported by Axon for example).

Example

Listing 1 demonstrates the initiation of the batch process.

BatchService allows for creating, updating, and retrieving the status of a batch.

The command message’s metadata is populated with the batch identifier, allowing it to be transparently passed to the subsequent event message. In Axon this may be enabled via the SimpleCorrelationDataProvider (listing 2).

Use this technique to make existing commands and events "batch-aware".

 
// create a new batch for registering a set of bikes.
var noBikes = bikeIds.length;
var batchId = batchService.createBatch(noBikes);

// register requested number of bikes in parallel.
//
// Note: The aggregate / command handler does not have to be 
// changed to accommodate batch processing.
for (var i = 0; i < noBikes; i++) {
    commandGateway.send(
        new RegisterBikeCommand(bikeIds[i]),
        MetaData.with("batchId", batchId)
    );
}

Listing 1 - Steps to initiate a batch process

// register a SimpleCorrelationDataProvider in Axon.
@Bean
public CorrelationDataProvider customCorrelationDataProvider() {
    return new SimpleCorrelationDataProvider("batchId");
}

Listing 2 - Ensure that the batch identifier is passed from the command to the event

BikeRegistered events are processed in Listing 3 as follows:

  • Retrieve the batch identifier (if any) from the event.

  • Update the batch status.

  • If all events that are part of the batch have been processed, continue with the next part of the flow.

@EventHandler
public void handle(
    BikeRegisteredEvent evt, 
    @MetaDataValue("batchId") String batchId
) {
    // update the RegisteredBikes read model...

    // retrieve the batch identifier from event metadata and 
    // increment the number of processed events.
    var status = batchService.updateBatch(batchId);
        
    if (status.noProcessed == status.expected) {
        // all events in this batch have been processed.
    }
}

Listing 3 - Update batch status and check if completion condition is fulfilled