Introducing Support for Bulk Updates in a Microservice (Legacy)

Important

Going forward, the Bulk Update flow is legacy code & will not receive new features. The support for Bulk Operations has been moved to Bulk Operations Orchestration service and is currently only available in Catalog. For this Broadleaf Common Bulk Operations library, all new domain for the Bulk Operations has been added to a v2 package. See the documentation for the v2 package.

Overview

The Broadleaf Common Bulk Operations library enables users to perform bulk operations on items in a data store. The library establishes the base configuration and default logic necessary to efficiently initiate and process these operations.This allows users of the library to focus on simply implementing the business-logic of each bulk operation rather than dealing with things like status management, batching, and error-reporting.

Basic Architectural Overview

At the heart of the bulk operations flow is the BulkUpdateManager. This is a service-level component that orchestrates the initiation and processing of each bulk operation. It contains boilerplate operation-agnostic logic to ensure each operation is initiated and processed correctly. Generally speaking, the DefaultBulkUpdateManager implementation should suffice for almost all use cases, and users should not need to extend or customize this unless there is a critical change required in how all operations are handled.

The BulkUpdateProcessor interface is the most important component that users need to implement in order to support a new bulk operation. The BulkUpdateManager injects a list of all BulkUpdateProcessor beans, and for each BulkUpdate, selects one to delegate operation-specific processing to.

Guide

Add a dependency on broadleaf-common-bulk-operations in your service’s pom.xml
```
<dependency>
  <groupId>com.broadleafcommerce.microservices</groupId>
  <artifactId>broadleaf-common-bulk-operations</artifactId>
</dependency>
```
1. This will automatically register the bulk operations auto configuration in your application via spring.factories.
Set values for essential properties pertaining to messaging configuration
1. Set values for channel message bindings. Here is a template:
  spring: cloud: stream: bindings: processBulkUpdateRequestInput: group: [your-service] // (1) destination: processBulkUpdateRequest // (2) processBulkUpdateRequestOutput: destination: processBulkUpdateRequest // (2) createSandboxRequestOutput: destination: createSandboxRequest deleteSandboxRequestOutput: destination: deleteSandboxRequest
  1. should be set to a value unique to your microservice to prevent more than one instance of your microservice from consuming a particular message.
  2. should also be a value that is unique to your microservice. For example, the Catalog microservice might set this value to processCatalogBulkUpdateRequest. This is because each ProcessBulkUpdateRequest is intended to be consumed by the same microservice (though not necessarily the same instance of that microservice) which produced it. Setting the channel to a value unique to the microservice ensures that other microservices are not unnecessarily sent ProcessBulkUpdateRequest messages that they cannot and should not process.
2. Set values for the durable notification configuration properties. These are the properties found in ProcessBulkUpdateRequestNotificationProperties, CreateSandboxRequestNotificationProperties, and DeleteSandboxRequestNotificationProperties. Generally speaking, these are good defaults:
  broadleaf: bulk: processBulkUpdateRequest: retry: retryClusterServiceNamespace: process-bulkupdate-request-[your-service-name] createSandboxRequest: retry: retryClusterServiceNamespace: create-sandbox-request-[your-service-name] deleteSandboxRequest: retry: retryClusterServiceNamespace: delete-sandbox-request-[your-service-name]

Implement BulkUpdateProcessor and register it as a bean

The BulkUpdateProcessor interface has three methods that must be implemented.

canHandle(BulkUpdate) is a method called by the BulkUpdateManager to select which BulkUpdateProcessor is used for a particular BulkUpdate.

Tip	Generally speaking, the result of this method should almost always be based upon the `BulkUpdate#target` and `BulkUpdate#type`. A new bulk update operation should have a new, distinct `target`/`type` combination.

Important

The first BulkUpdateProcessor for which this method returns true is used to process the bulk operation. The order in which these processors are checked is entirely based upon their bean ordering. To give a particular processor higher precedence, simply register its bean with a higher precedence value.

readRecordsToProcess(BulkUpdate) is a method called by the BulkUpdateManager to retrieve the full stream of base records that the bulk operation will be applied to.
modifyAndPersist(List<P>, BulkUpdate) is a method called by the BulkUpdateManager for each batch subset of the records returned by readRecordsToProcess(BulkUpdate).
1. This is the method that should be responsible for deserializing and interpreting the BulkUpdate#serializedPayload (which should be specific for the BulkUpdate#target/BulkUpdate#type combination), using it to apply the necessary changes for each record in the given batch, and then persisting the results.
  1. While readRecordsToProcess(BulkUpdate) can only filter and find records of a particular domain, this method is not restricted to only modifying those records. Implementors may decide to query for and update records in other related domains within this method.
2. This method is invoked repeatedly by the BulkUpdateManager until there are no more records in the full stream.
3. The maximum size of the List<P> batch taken from the stream and passed as an argument to this method is configurable via the broadleaf.bulk.processing.batchSize configuration property defined in BulkUpdateProcessingConfigurationProperties. By default, it is 100. Increasing this value to a larger number may speed up processing at the cost of increased memory usage.

Add endpoints to accept new bulk operation requests and expose the status of previously requested operations.

Tip	These should be very simple endpoints.

Add an endpoint to accept new requests for performing bulk operations.

The ultimate goal of this endpoint is to call BulkUpdateManager#initiateBulkUpdate() with all of the necessary information, and let it handle the rest. Here is an example endpoint implementation:

@FrameworkPostMapping(value = "/bulk-updates", consumes = MediaType.APPLICATION_JSON_VALUE)
public void requestBulkUpdateForMyTargetEntity(
        @RequestParam(value = "cq", defaultValue = "") String filterString,
        @RequestBody BulkUpdateRequest bulkUpdateRequest,
        @ContextOperation(value = OperationType.UPDATE) ContextInfo context) {
    bulkUpdateManager.initiateBulkUpdate(
        bulkUpdateRequest,
        "MY BULK UPDATE TARGET ENTITY",
        filterString,
        context);
}

Add an endpoint to return the information about previously requested BulkUpdate operations.

This is purely intended to allow users to see the statuses and progress of their previously requested bulk operations. Here is an example endpoint implementation that uses BulkUpdateService#readByTarget():

@FrameworkGetMapping(value = "/bulk-updates")
public Page<BulkUpdate> getBulkUpdatesForMyTargetEntity(
        @ContextOperation(value = OperationType.READ) ContextInfo context,
        @PageableDefault(size = 50) Pageable pageable) {
    return bulkUpdateService.readByTarget(
        "MY BULK UPDATE TARGET ENTITY",
        context,
        pageable);
}

Introduce metadata to allow requesting bulk operations from the admin

Since bulk operations are always in the context of a particular entity/target, the metadata for bulk operations is configured via EntityGridBuilder#configureBulkUpdate(). This method should be called on the entity grid for the domain that will be the target of the bulk operation. The arguments to this configuration should match the setup of your endpoints from the previous step.
1. The Consumer<BulkUpdateActionBuilder> bulkUpdateConfigurer function should add the following configuration:
  1. BulkUpdateActionBuilder#configureGrid() allows customization of the grid that will display the previously-requested bulk operations. Within it, BulkUpdateGridBuilder.configureRead() should be configured to fetch the previously requested bulk operations via the endpoint created in the previous step.
  2. BulkUpdateActionBuilder.configureTargetGrid() is used to configure the grid that admin users will use to filter and find records that the bulk operation will target. This is ultimately used to determine the values for BulkUpdate#inclusions, BulkUpdate#exclusions, and BulkUpdate#filterString. Use this configurer to set the correct endpoint that can be used to find and filter the target domain records, and then customize the grid that will display those results to the admin user.
  3. BulkUpdateActionBuilder.configureForm() is used to configure all of the different types (BulkUpdate#type) the user can select from, and for each type defines the form that will be used to create the BulkUpdate#serializedPayload. Within this configurer, SelectTargetEntityFormBuilder.addOperation() is used to register each type and its corresponding form components. Ensure to register the custom types and forms as required by your BulkUpdateProcessor implementation.