Correlation Set in BPEL to Interrupt BPEL

Its been quite sometime since I blogged last – got very busy for past few months. Back now with something that I feel interesting.

Use case: A composite instance triggers a BPEL component. The BPEL component may invoke many asynchronous flows or it might be in wait state using “wait” activity or it might use a Human task and wait till human task completes. These scenarios might dehydrate BPEL. In this case, I take a scenario where I want to terminate the BPEL component instance when BPEL process is waiting for some legitimate reason.

Technical solution: The composite exposes a service with two operations. One MainOperation that would initiate the main flow of the BPEL. Two InterruptOperation that would trigger the OnMessage event handler of the same BPEL process and terminate the corresponding component instance initiated through MainOperation using Correlation sets.

Sample app used below: http://kavin-sample-apps.googlecode.com/files/InterruptBPEL.rar

Description of the project:

Create a SOA Application and Project with an empty SOA Composite.

The input schema that I have chosen for this composite is below. This schema has two messages.

1) EmployeeDetails -> Detail sent to initiate the main BPEL flow which has the core business logic. EmployeeId in this message is used in correlation set.

2) InterruptPayload -> This is used by InterruptOperation that is used to terminate the BPEL instance. This also carries EmployeeId so that the main flow can be correlated.

Create a new BPEL Process. Choose Template as “Base on a WSDL” and create a new WSDL file. Choose an one way operation and name it as “EmployeeProcessingFlow”. This is going to be the operation that is going to execute the core business logic in the BPEL process. This operation would be using “EmployeeDetails” element in above schema.

Instead of creating a new composite service for the Interrupt flow, let us add an additional operation to the existing WSDL in the same port. This operation is going to use a newly defined WSDL message “InterruptMessage” that would point to “InterruptPayload” in above given schema.

<wsdl:message name="requestMessage">
<wsdl:part name="EmployeeDetailMessage" element="inp1:EmployeeDetails"/>
</wsdl:message>
<wsdl:message name="InterruptMessage">
<wsdl:part name="InterruptMessage" element="inp1:InterruptPayload"/>
</wsdl:message>
<wsdl:portType name="EmployeeProcessingPort">
<wsdl:operation name="EmployeeProcessingFlow">
<wsdl:input message="tns:requestMessage"/>
</wsdl:operation>
<wsdl:operation name="InterruptFlow">
<wsdl:input message="tns:InterruptMessage"/>
</wsdl:operation>
</wsdl:portType>

Open the newly created BPEL process – “EmployeeProcessingProcess.bpel”. Add an OnMessage Event Handler to the global scope. This branch is going to wait for the incoming request from “InterruptFlow” operation of “EmployeeProcessingService”.

Now let’s correlate the receive and OnMessage entry point of EmployeeProcessingProcess into a single flow using correlation sets. To know more about correlation sets, refer to this section of Oracle SOA Suite Developers Guide. Basically correlation set is a way to attach a business specific key information to a BPEL instance so that the specific instance can be fetched at runtime and process can be executed further.

To start with, edit the receive activity of the BPEL process and add a new Correlation. Create a new Correlation set – “EmployeeFlowCorrelationSet”  with a property to bind the business object specific unique key – BindEmployeeId. Add the EmployeeId element from the Input message of the receive activity as a “Property Alias” for Property. Make sure “Initiate” property of this correlation in this receive activity to “yes”.

Tip: In above screenshot, NOTE that there is no icon to “choose” Query declaratively. So point the cursor there and press Ctrl+Enter so that it would show the message automatically to select.

Add a wait activity after receive and set wait time to 5 minutes to simulate the use case. On OnMessage branch, do a terminate to cancel the BPEL flow. So when the first message comes from “EmployeeProcessingFlow”, the BPEL will be in wait state for 5minutes. In the interim, when message from “InterruptFlow” comes, then it would be correlated with old instance and the instance would be terminated. If certain clean up need to be done before termination, then it can be included in OnMessage branch before Terminate.

Edit the OnMessage branch and bind the same correlation. In this OnMessage branch, the EmployeeId from InterruptMessage has to be set as Property Alias for the existing correlation set.

That completes the coding 🙂

Testing the project:

Now deploy the composite to EM and access the end point using the default end point tester URL.

http://localhost:7001/soa-infra/services/default/InterruptBpelFlow/EmployeeProcessingService_ep

Select the operation “EmployeeProcessingFlow” that handles main flow. Enter an EmployeeId value like “349827495” and a random Salary value and Invoke this to Initiate the composite main flow. This would initiate the BPEL process receive activity by setting EmployeeId as correlation id and put the BPEL in wait state for 5 minutes which can be seen from EM.

Now on the same URL [http://localhost:7001/soa-infra/services/default/InterruptBpelFlow/EmployeeProcessingService_ep], select the “InterruptFlow” operation with same EmployeeId as input so that it would correlate with above mentioned BPEL instance and terminate it.

Thanks for Reading and Keep Smiling 🙂

Kavin