JBoss Fuse and JBoss BPM Suite Integrated

Last week Apache Camel v2.16 was released with over 600 issues resolved (bug fixes, updates, and new features). Claus blogged about top 10 highlights already and here I want to briefly show you a use case about my tiny contribution in this release: camel-jbpm component.

This component is intended to make integration with jBPM (a Business Process Management Application) easier. It uses Remote Java API and basically does REST calls to jBPM to manage business processes (start, stop, abort), execute Business Rules, manage Human Tasks, etc. So rather than manually creating a REST client or the Remote API, you can use the Camel component. The remote API provides also HornetMQ based integration, but I've dropped out that part from the Camel component as it doesn't play nicely with OSGI.

In the sample use case (which is inspired by a real one) there is an integration application based on Camel that runs on JBoss Fuse which is supposed to do real time integration. And whenever an error occurs in the integration route, the error details are sent to jBPM for evaluation and review by human being if necessary. To demonstrate that I've created two separate projects:

• A Camel project that runs on Fuse, handles any exception and sends the details to jBPM
• A Business Process project that runs on jBPM, classifies the errors using Drools and Complex Event Processing and creates Human Tasks if the error is critical.

The Camel route below is self explanatory, there is a route that generates 10 exceptions with half a second delay each and an error handler that catches those exceptions and sends them to jBPM. Notice that it also populates CamelJBPMParameters header with the details about the error.

public void configure() {

    from("timer://trigger?fixedRate=true&period=500&repeatCount=10").routeId("mainRoute")
            .to("log:com.ofbizian.jbpm.before?showAll=true&multiline=true")
            .process(new Processor() {
                @Override
                public void process(Exchange exchange) throws Exception {
                    throw new RuntimeException("Something went wrong");
                }
            });

    onException(Exception.class)
            .handled(true)
            .process(new Processor() {
                @Override
                public void process(Exchange exchange) throws Exception {
                    final Map map = new HashMap();
                    map.putAll(exchange.getIn().getHeaders());
                    map.put("contextId", exchange.getContext().getName());
                    map.put("routeId", exchange.getProperty(Exchange.FAILURE_ROUTE_ID));
                    map.put("endpointId", exchange.getProperty(Exchange.FAILURE_ENDPOINT));
                    map.put("exchangeId", exchange.getExchangeId());
                    map.put("breadcrumbId", exchange.getIn().getHeader(Exchange.BREADCRUMB_ID));
                    map.put("exceptionType", exchange.getProperty(Exchange.EXCEPTION_CAUGHT).getClass());
                    map.put("errorMessage", exchange.getProperty(Exchange.EXCEPTION_CAUGHT).toString());
                    exchange.getOut().setHeader("CamelJBPMParameters", map);
                }
            })
            .to("jbpm:http://localhost:8080/business-central?userName=bpmsAdmin&password=password&deploymentId=org.kie.example:camel-process:1.0.0-SNAPSHOT&processId=project1.camel.demo");
}   

You can find the sample code on github with instructions about how to build and run it on Fuse.

The second part of the demo consist of the jBPM process that receives the errors and processes them. The BPMN2 diagram of the process is below:

The process have input parameters matching the values from CamelJBPMParameters header. In the first step it initializes a RequestQualifier object that is marked as non-critical. In the second step we use Drools and Complex Event Processing to qualify errors that happen over a period of time. The idea with CEP is that we want to mark an error as critical only if there were more than 5 errors in 10 seconds time window. Also to prevent marking as critical multiple errors and flooding the system with Human Tasks, we limit to only 1 critical error at most in 10 minutes interval. All these conditions are written in Drools as following:

package org.kie.example.project1;

rule "Mark a request as critical when there are 5 or more in 10 seconds. There should be at most 1 critical per 10 minutes"
when
    $requestQualifier: RequestQualifier(processed == false)
    Number( intValue >=5 ) from accumulate($r: RequestQualifier( ) over window:time( 10s ),count( $r ) )
    Number( intValue ==0 ) from accumulate($r: RequestQualifier( critical == true) over window:time(10m ),count( $r ) )
then
    modify($requestQualifier) {
        setProcessed(true),
        setCritical(true);
    }
end

Once the request with the error details has been qualified as critical or not, we use that information to do conditional routing. All non critical errors are logged and the business process completes. We can see the full list of completed processes and error related information in the following screenshot:

There is no further action required for those business process instances. But in the other branch of the business process, if an error has been marked as critical, we do create a Human Task and the process stops at this point.

This requires a user to assign the task and take actions on it. We can see that the Human Task has received the details from the business process:

Once the Human task is claimed and completed, then the business process continues and completes in the next step.