Chapter 25. Concurrency and asynchronous execution

public class MyServiceTaskHandler implements WorkItemHandler {

        
  public void executeWorkItem(WorkItem workItem, WorkItemManager manager) {
    new Thread(new Runnable() {
      public void run() {
        // Do the heavy lifting here ...
      }
    }).start();
  }

  public void abortWorkItem(WorkItem workItem, WorkItemManager manager) {
  }
}

It's advisable to have your handler contact a service that executes the business operation, instead of having it perform the actual work. If anything goes wrong with a business operation, it doesn't affect your process. The loose coupling that this provides also gives you greater flexibility in reusing services and developing them.

For example, you can have your human task handler simply invoke the human task service to add a task there. To implement an asynchronous handler, you usually have to simply do an asynchronous invocation of this service. This usually depends on the technology you use to do the communication, but this might be as simple as asynchronously invoking a web service, or sending a JMS message to the external service.

25.2.2. jbpm executor

In version 6, jBPM introduces new component called jbpm executor which provides quite advanced features for asynchronous execution. It delivers generic environment for background execution of commands. Commands are nothing more than business logic encapsulated within simple interface. It does not have any process runtime related information, that means no need to complete work items, or anything of that sort. It purely focuses on the business logic to be executed. It receives data via CommandContext and returns results of the execution with ExecutionResults.

Before looking into details on jBPM support for asynchronous execution let's look at what are the common requirements for such execution:

allows asynchronous execution of given piece of business logic
allows to retry in case of resources are temporarily unavailable e.g. external system interaction
allows to handle errors in case all retries have been attempted
provides cancellation option
provides history log of execution

When confronting these requirements with the "simple async handler" (executed as separate thread) you can directly notice that all of these would need to be implemented all over again by different systems. Due to that a common, generic component has been provided out of the box to simplify and empower usage.

jBPM executor operates on commands, which are essential piece of code that is going to be executed as background job.

/**
 * Executor's Command are dedicated to contain purely business logic that should be executed. 
 * It should not have any reference to underlying process engine and should not be concerned
 * with any process runtime related logic such us completing work item, sending signals, etc.
 * <br/>
 * Information that are taken from process will be delivered as part of data instance of 
 * <code>CommandContext</code>. Depending on the execution context that data can vary but 
 * in most of the cases following will be given:
 * <ul>
 *  <li></li>
 *  <li>businessKey - usually unique identifier of the caller</li>
 *  <li>callbacks - FQCN of the <code>CommandCollback</code> that shall be used on command completion</li>
 * </ul>
 * When executed as part of the process (work item handler) additional data can be expected:
 * <ul>
 *  <li>workItem - the actual work item that is being executed with all its parameters</li>
 *  <li>processInstanceId - id of the process instance that triggered this work</li>
 *  <li>deploymentId - if given process instance is part of an active deployment</li>
 * </ul>
 * Important note about implementations is that it shall always be possible to be initialized with default constructor
 * as executor service is an async component so it will initialize the command on demand using reflection.
 * In case there is a heavy logic on initialization it should be placed in another service implementation that 
 * can be looked up from within command.
 */
public interface Command {
    
    /**
     * Executed this command's logic.
     * @param ctx - contextual data given by the executor service
     * @return returns any results in case of successful execution
     * @throws Exception in case execution failed and shall be retried if possible
     */
    public ExecutionResults execute(CommandContext ctx) throws Exception;
}

Looking at the interface above, there is no specific integration with the jBPM runtime engine, it's decoupled from it to put main focus on the actual logic that shall be executed as part of that command rather to worry about integration with process engine. This design promotes reuse of already existing logic by simply wrapping it with Command implementation.

Input data is transferred from process engine to command via CommandContext. It acts purely as data transfer object and puts single requirement on the data it holds - all objects must be serializable.

/**
 * Data holder for any contextual data that shall be given to the command upon execution.
 * Important note that every object that is added to the data container must be serializable 
 * meaning it must implement <code>java.io.Seriazliable</code>
 *
 */
public class CommandContext implements Serializable {

    private static final long serialVersionUID = -1440017934399413860L;
    private Map<String, Object> data;

    public CommandContext() {
        data  = new HashMap<String, Object>();
    }

    public CommandContext(Map<String, Object> data) {
        this.data = data;
    }

    public void setData(Map<String, Object> data) {
        this.data = data;
    }

    public Map<String, Object> getData() {
        return data;
    }

    public Object getData(String key) {
        return data.get(key);
    }

    public void setData(String key, Object value) {
        data.put(key, value);
    }

    public Set<String> keySet() {
        return data.keySet();
    }

    @Override
    public String toString() {
        return "CommandContext{" + "data=" + data + '}';
    }
}

Next outcome is provided to process engine via ExecutionResults, which is very similar in nature to the CommandContext and acts as data transfer object.

/**
 * Data holder for command's result data. Whatever command produces should be placed in
 * this results so they can be later on referenced by name by the requester - e.g. process instance.
 *
 */
public class ExecutionResults implements Serializable {

    private static final long serialVersionUID = -1738336024526084091L;
    private Map<String, Object> data = new HashMap<String, Object>();

    public ExecutionResults() {
    }

    public void setData(Map<String, Object> data) {
        this.data = data;
    }

    public Map<String, Object> getData() {
        return data;
    }

    public Object getData(String key) {
        return data.get(key);
    }

    public void setData(String key, Object value) {
        data.put(key, value);
    }

    public Set<String> keySet() {
        return data.keySet();
    }

    @Override
    public String toString() {
        return "ExecutionResults{" + "data=" + data + '}';
    }
    
    
}

Executor covers all requirements listed above and provides user interface as part of jbpm console and kie workbench (kie-wb) applications.

Figure 25.1.

Above screenshot illustrates history view of executor's job queue. As can be seen on it there are several options available:

view details of the job
cancel given job
create new job


/**
 * Marks given executor command it is reoccurring and shall be rescheduled after completion of single instance.
 *
 */
public interface Reoccurring {

	/**
	 * Returns next time to be scheduled. Date must be in future as jobs cannot be scheduled in past.
	 * Returns null in case it should not be scheduled any more.
	 * @return
	 */
	Date getScheduleTime();
}

Reoccurring interface is very simple and enforces implementation to provide the next schedule time that the command should be executed at. It must already be valid date that is not in the past. In case no more invocation of given command should happen return value of this method should be null.

An excellent example of such command is org.jbpm.executor.commands.LogCleanupCommand that provides easy and convineint way to schedule periodic clean up of jBPM log tables on defined time intervals. See this article to see it in action and how to configure and run it.

25.2.2.4. Run jobs on same server node on which it was scheduled

By default jbpm executor is cluster ready and by that will distribute jobs across all cluster members. That might result in execution of given job on different cluster member than it was scheduled which is not always desired. To override this mechanism job can set 'Owner' as part of their data when being registered wher owner is the executor instance that is scheduling the job.


 CommandContext ctx = new CommandContext();
 ctx.setData("some data", "data...");
 ctx.setData("retries", 0);
 ctx.setData("owner", ExecutorService.EXECUTOR_ID);

That will ensure that only the isntance that scheduled the job will be the one which will execute it. Note that it might impact the time when the job is executed especially in cases where given cluster member will be unavailable.

25.2.2.5. Assign priority to jobs

Asynchronous jobs are by default executed based on their scheduled time thus in case several jobs are scheduled to be executed at the same time there might be an issue with which one will be executed first. To override default behavior priorities can be assigned to individual jobs. Priority is given as integer from 0-9 range where 0 is the lowest priority and 9 is the highest. This can be done:

directly via CommandContext using priority context data object where value is a valid integer from 0-9 range
via data inputs of Async task where data input property name is Priority and value is a valid integer from 0-9 range

With assigned priority jBPM executor will pick the jobs based on the scheduled time and their priority. In case there is no priority assigned jBPM executor will set it to 5 as default.

jBPM executor can utilize JMS broker for notifying about job to be executed (only jobs to be executed immediately) and the priority then is also set on JMS message so the JMS broker will take this into account on delivery.