Managing, Purging and Cleanup of the BPEL Instances (2)

As promised in my last post, I am posting the bpel_dehyd_manage.sql script. The script will alter the tables, deallocate unused data space, alter the indexes and call the purge_instances procedure.

set echo onset time onset timing on
spool bpel_dehyd_manage.log

-- Call the purge instance procdure to truncate the data from the BPEL Instances tables
exec purge_instances(SYSDATE-&1);

-- The following statements reclaims the deleted data space
alter table cube_instance deallocate unused;
alter table cube_scope deallocate unused;
alter table work_item deallocate unused;
alter table wi_exception deallocate unused;
alter table document_ci_ref deallocate unused;
alter table document_dlv_msg_ref deallocate unused;
alter table scope_activation deallocate unused;
alter table dlv_subscription deallocate unused;
alter table audit_trail deallocate unused;
alter table audit_details deallocate unused;
alter table sync_trail deallocate unused;
alter table sync_store deallocate unused;
alter table dlv_message deallocate unused;
alter table invoke_message deallocate unused;
alter table ci_indexes deallocate unused;

alter table cube_scope enable row movement;
alter table cube_scope shrink space compact;
alter table cube_scope shrink space;
alter table cube_scope disable row movement;

alter table cube_instance enable row movement;
alter table cube_instance shrink space compact;
alter table cube_instance shrink space;
alter table cube_instance disable row movement;

alter table work_item enable row movement;
alter table work_item shrink space compact;
alter table work_item shrink space;
alter table work_item disable row movement;

alter table wi_exception enable row movement;
alter table wi_exception shrink space compact;
alter table wi_exception shrink space;
alter table wi_exception disable row movement;

alter table document_ci_ref enable row movement;
alter table document_ci_ref shrink space compact;
alter table document_ci_ref shrink space;
alter table document_ci_ref disable row movement;

alter table document_dlv_msg_ref enable row movement;
alter table document_dlv_msg_ref shrink space compact;
alter table document_dlv_msg_ref shrink space;
alter table document_dlv_msg_ref disable row movement;

alter table scope_activation enable row movement;
alter table scope_activation shrink space compact;
alter table scope_activation shrink space;
alter table scope_activation disable row movement;

alter table dlv_subscription enable row movement;
alter table dlv_subscription shrink space compact;
alter table dlv_subscription shrink space;
alter table dlv_subscription disable row movement;

alter table audit_trail enable row movement;
alter table audit_trail shrink space compact;
alter table audit_trail shrink space;
alter table audit_trail disable row movement;

alter table audit_details enable row movement;
alter table audit_details shrink space compact;
alter table audit_details shrink space;
alter table audit_details disable row movement;

alter table sync_trail enable row movement;
alter table sync_trail shrink space compact;
alter table sync_trail shrink space;
alter table sync_trail disable row movement;

alter table sync_store enable row movement;
alter table sync_store shrink space compact;
alter table sync_store shrink space;
alter table sync_store disable row movement;

alter table invoke_message enable row movement;
alter table invoke_message shrink space compact;
alter table invoke_message shrink space;
alter table invoke_message disable row movement;

alter table dlv_message enable row movement;
alter table dlv_message shrink space compact;
alter table dlv_message shrink space;
alter table dlv_message disable row movement;

alter table ci_indexes enable row movement;
alter table ci_indexes shrink space compact;
alter table ci_indexes shrink space;
alter table ci_indexes disable row movement;

alter table XML_DOCUMENT enable row movement;
alter table XML_DOCUMENT shrink space compact;
alter table XML_DOCUMENT shrink space;
alter table XML_DOCUMENT disable row movement;

-- Rebuild the indexes of the ORABPEL schema for faster lookups
spool off
set termout off
set heading off
set echo off
set time off
set timing off

spool rebuild_indexes.sql

select 'Alter Index 'index_name' Rebuild;' from from user_indexes where table_name not like 'BIN$%' and index_type <> 'LOB';

spool off

set termout on
set echo on

spool rebuild_indexes.log
@rebuild_indexes.sql

spool off
spool off

The above table will manage you BPEL instances and you may schedule this script run based on your organization or client requirement.

There is one more tip for the dehydation store of the BPEL processes - move the LOB Columns to non assm tablespaces to eliminate contention issues and allow indexing for the CUBE_SCOPE,
AUDIT_DETAILS, SYNC_STORE and XML_DOCUMENT tables in the ORABPEL schema. You may a blog entry on how to perform the above operation,

http://scsoablog.blogspot.com/2008/02/more-hw-contention-tuning.html
http://scsoablog.blogspot.com/2008/01/optimizing-rac-performance-for-soa-part.html

Cheers!

Managing, Purging and Cleanup of the BPEL Instances (1)

The customers dealing with high volume of BPEL messages can see a impact on the performance of the processes as the database (ORABPEL) grows in size. The reasons are obvious, time-consuming database lookups , contention issues and so on. In order to peform the scheduled cleanup of the ORABPEL schema, create a procedure in the database as under,

CREATE OR REPLACE PROCEDURE purge_instances (p_older_than TIMESTAMP)AS
BEGIN

--Before starting clean up of temp tables
EXECUTE IMMEDIATE 'TRUNCATE TABLE temp_cube_instance';
EXECUTE IMMEDIATE 'TRUNCATE TABLE temp_invoke_message';
EXECUTE IMMEDIATE 'TRUNCATE TABLE temp_dlv_message';

--Populate table with information about older instances
INSERT into temp_cube_instance

SELECT cikey
FROM cube_instance
WHERE state >= 3 AND modify_date < p_older_than;

INSERT into temp_invoke_message
SELECT message_guid
FROM invoke_message
WHERE state > 1 AND receive_date < p_older_than;

INSERT into temp_dlv_message
SELECT message_guid
FROM dlv_message
WHERE state > 1 AND receive_date < p_older_than;
COMMIT;

-- WHERE clause is to force CBO to use index of temp table and to avoid full scan of temp
table
-- COMMIT after every delete to reduce pressure on undo log in database

-- Delete all closed instances older than specified date

DELETE FROM native_correlation
WHERE conversation_id IN (SELECT /*+ ORDERED */ dlvs.conv_id FROM dlv_subscription dlvs,
temp_cube_instance tpic
WHERE dlvs.cikey = tpic.cikey );
COMMIT;

DELETE FROM cube_scope
WHERE cikey IN (SELECT /*+ ORDERED */ cs.cikey FROM cube_scope cs, temp_cube_instance tpic
WHERE cs.cikey = tpic.cikey);
COMMIT;

DELETE FROM work_item
WHERE cikey IN (SELECT /*+ ORDERED */ wi.cikey FROM work_item wi, temp_cube_instance tpic
WHERE wi.cikey = tpic.cikey);
COMMIT;

DELETE FROM wi_exception
WHERE cikey IN (SELECT /*+ ORDERED */ wie.cikey FROM wi_exception wie,
temp_cube_instance tpic
WHERE wie.cikey = tpic.cikey);
COMMIT;

DELETE FROM scope_activation
WHERE cikey IN (SELECT /*+ ORDERED */ sa.cikey FROM scope_activation sa,
temp_cube_instance tpic
WHERE sa.cikey = tpic.cikey);
COMMIT;

DELETE FROM dlv_subscription
WHERE cikey IN (SELECT /*+ ORDERED */ dlvs.cikey FROM dlv_subscription dlvs,
temp_cube_instance tpic
WHERE dlvs.cikey = tpic.cikey);
COMMIT;

DELETE FROM audit_trail
WHERE cikey IN (SELECT /*+ ORDERED */ at.cikey FROM audit_trail at, temp_cube_instance tpic
WHERE at.cikey = tpic.cikey);
COMMIT;

DELETE FROM audit_details
WHERE cikey IN (SELECT /*+ ORDERED */ ad.cikey FROM audit_details ad,
temp_cube_instance tpic
WHERE ad.cikey = tpic.cikey);
COMMIT;

DELETE FROM sync_trail
WHERE cikey IN (SELECT /*+ ORDERED */ st.cikey FROM sync_trail st, temp_cube_instance tpic
WHERE st.cikey = tpic.cikey);
COMMIT;

DELETE FROM sync_store
WHERE cikey IN (SELECT /*+ ORDERED */ ss.cikey FROM sync_store ss, temp_cube_instance tpic
WHERE ss.cikey = tpic.cikey);
COMMIT;

DELETE FROM xml_document
WHERE dockey IN (SELECT /*+ ORDERED */ doc_ref.dockey FROM document_ci_ref doc_ref,
temp_cube_instance tpic
WHERE doc_ref.cikey = tpic.cikey);
COMMIT;

DELETE FROM document_dlv_msg_ref
WHERE dockey IN (SELECT /*+ ORDERED */ doc_ref.dockey FROM document_ci_ref doc_ref,
temp_cube_instance tpic
WHERE doc_ref.cikey = tpic.cikey);
COMMIT;

DELETE FROM document_ci_ref
WHERE cikey IN (SELECT /*+ ORDERED */ dcr.cikey FROM document_ci_ref dcr,
temp_cube_instance tpic
WHERE dcr.cikey = tpic.cikey);
COMMIT;

DELETE FROM attachment
WHERE key IN (SELECT /*+ ORDERED */ attach_ref.key FROM attachment_ref attach_ref,
temp_cube_instance tpic
WHERE attach_ref.cikey = tpic.cikey);
COMMIT;

DELETE FROM attachment_ref
WHERE cikey IN (SELECT /*+ ORDERED */ ar.cikey FROM attachment_ref ar,
temp_cube_instance tpic
WHERE ar.cikey = tpic.cikey);
COMMIT;

DELETE FROM ci_indexes
WHERE cikey IN (SELECT /*+ ORDERED */ cin.cikey FROM ci_indexes cin,
temp_cube_instance tpic
WHERE cin.cikey = tpic.cikey);
COMMIT;

-- DELETE FROM wi_fault
-- WHERE cikey IN (SELECT /*+ ORDERED */ wf.cikey FROM wi_fault wf,
-- temp_cube_instance tpic
-- WHERE wf.cikey = tpic.cikey);
-- COMMIT;

DELETE FROM cube_instance
WHERE cikey IN (SELECT /*+ ORDERED */ ci.cikey FROM cube_instance ci,
temp_cube_instance tpic
WHERE ci.cikey = tpic.cikey);
COMMIT;

-- Purge all handled invoke_messages older than specified date --
DELETE FROM xml_document
WHERE dockey IN (SELECT /*+ ORDERED */ dlv_ref.dockey FROM document_dlv_msg_ref dlv_ref,
temp_invoke_message tpiim
WHERE dlv_ref.message_guid = tpiim.message_guid);
COMMIT;

DELETE FROM document_ci_ref
WHERE dockey IN (SELECT /*+ ORDERED */ dlv_ref.dockey FROM document_dlv_msg_ref dlv_ref,
temp_invoke_message tpiim
WHERE dlv_ref.message_guid = tpiim.message_guid);
COMMIT;

DELETE FROM document_dlv_msg_ref
WHERE message_guid IN (SELECT /*+ ORDERED */ dlv_ref.message_guid FROM document_dlv_msg_ref dlv_ref,
temp_invoke_message tpiim
WHERE dlv_ref.message_guid = tpiim.message_guid);
COMMIT;

DELETE FROM invoke_message
WHERE message_guid IN (SELECT /*+ ORDERED */ im.message_guid FROM invoke_message im,
temp_invoke_message tpiim
WHERE im.message_guid = tpiim.message_guid);
COMMIT;

-- Purge all handled callback messages older than specified date --
DELETE FROM xml_document
WHERE dockey IN (SELECT /*+ ORDERED */ dlv_ref.dockey FROM document_dlv_msg_ref dlv_ref,
temp_dlv_message tpidm
WHERE dlv_ref.message_guid = tpidm.message_guid);
COMMIT;

DELETE FROM document_ci_ref
WHERE dockey IN (SELECT /*+ ORDERED */ dlv_ref.dockey FROM document_dlv_msg_ref dlv_ref,
temp_dlv_message tpidm
WHERE dlv_ref.message_guid = tpidm.message_guid);
COMMIT;

DELETE FROM document_dlv_msg_ref
WHERE message_guid IN (SELECT /*+ ORDERED */ ddmr.message_guid FROM document_dlv_msg_ref ddmr,
temp_dlv_message tpidm
WHERE ddmr.message_guid = tpidm.message_guid);
COMMIT;

DELETE FROM dlv_message
WHERE message_guid IN (SELECT /*+ ORDERED */ dm.message_guid FROM dlv_message dm,
temp_dlv_message tpidm
WHERE dm.message_guid = tpidm.message_guid);
COMMIT;

-- delete all unreferenced xml_documents rows from xml_document table
DELETE FROM xml_document xd
WHERE NOT EXISTS (SELECT ddmr.dockey FROM document_dlv_msg_ref ddmr
WHERE xd.dockey = ddmr.dockey) AND NOT EXISTS (SELECT dir.dockey FROM document_ci_ref dir WHERE xd.dockey = dir.dockey);
COMMIT;

-- IF conversation_id is not present in dlv_subscription, we can delete it from native_correlation
DELETE FROM native_correlation nc
WHERE NOT EXISTS (SELECT dlvs.conv_id from dlv_subscription dlvs
WHERE dlvs.conv_id = nc.conversation_id);
COMMIT;

DELETE FROM process_log
WHERE event_date < p_older_than;

EXECUTE IMMEDIATE 'TRUNCATE TABLE temp_cube_instance';
EXECUTE IMMEDIATE 'TRUNCATE TABLE temp_invoke_message';
EXECUTE IMMEDIATE 'TRUNCATE TABLE temp_dlv_message';
COMMIT;

END purge_instances;

The above procedure shall truncate all the BPEL instance tables, but in order to reclaim the dataspace and refresh the indexes and you should run the another script (bpel_dehyd_manage.sql). I will publish the script in the next post.

You can run the above procedure as, call purge_instances(sysdate - ).

How to access the SOAP/Custom Headers in BPEL/ESB or pass the SOAP/Custom Headers from BPEL/ESB?

SOAP headers is a way to pass the data to and from an XML Web service method if the data is not directly related to the XML Web service method's primary functionality. They are primarily used to pass authentication details or any non-functional data. Since BPEL specification also forces to have a interface definition and communication via SOAP, these headers can be leveraged for various non-functional requirements.

How to access the SOAP Headers in BPEL Process or pass the SOAP Headers from BPEL Process?

You need to make some simple modifications to some of the files for accessing the information passed in the SOAP Headers and the steps are as under,

1. Add the definition of the header variable expected in the soap:Header in [process_name].xsd or add the XSD to the BPEL project as under,
   <element name="Header">
   <complexType>
   <sequence>
   <element name="Data" type="string"/>
   </sequence>
   </complexType>
   </element>
   
   
2. Add the namespace, if absent to the [process].wsdl
   xmlns:soap="http://schemas.xmlsoap.org/wsdl/soap/"
   
   
3. Add the message type of the header variable (passed in the soap:Header) in the [process].wsdl
   <message name="Header">
   <part name="Header" element="ns1:Header"/>
   </message>
   Note: ns1 is the namespace of the header variable as mentioned in step1
   
   
4. Add the Header variable under the variable element in the [process].bpel as under,
   <variable name="Header" messageType="ns1:Header"/>
   
   
5. Add the attribute bpelx:headerVariable to the recieve activity element as under,
   <receive name="receiveInput" partnerLink="client"
   portType="client:SyncBPELProcess" operation="initiate"
   variable="inputVariable" createInstance="yes"
   bpelx:headerVariable="Header"/>

Make sure that the namespace declaration of the Header Variable is available in .bpel and .wsdl file.

After making the above mentioned modification to the BPEL Process, you need to manually create the soap:Header if you are using a UI tool for testing the process or develop another BPEL Process that shall call the modified BPEL Process and populate the SOAP Header. Both the ways are simple and steps are as under,

• Manual: Add/Modify the <soap:Header> element to include the Header variable as under,
  <soapenv:Header
  xmlns:soapenv="http://schemas.xmlsoap.org/
  soap/envelope/"
  xmlns:asy="http://xmlns.oracle.com/SyncBPELProcess">
  <asy:Header>
  <asy:Data>HEADER DATA</asy:Data>
  </asy:Header>
  </soapenv:Header>
  
  
• Pass SOAP Header from BPEL Process: Create a new BPEL project (CallBPELProj) and import the definition of the Header Variable in the project. Create a variable(HeaderVariable) of type Header variable in the BPEL project and invoke the Header BPEL project (modified above) and add the attribute bpelx:inputHeaderVariable to the invoke element as under,
  <invoke name="Invoke_1" partnerLink="AsyncBPELProcess"
  portType="ns1:AsyncBPELProcess" operation="initiate"
  inputVariable="Invoke_1_initiate_InputVariable"
  bpelx:inputHeaderVariable="HeaderVariable"/>
  

With all these changes, you shall be able to access the information passed in the SOAP Headers.

There are some pre-defined Header variable, as mentioned in WS-Addressing specification that are passed as SOAP Header for implementing the callback and setting the message destinations in case of Asynchronous invocations. The above modification will work fine in all the cases,

• WS-Client passing the Custom Header variable to BPEL Process(interface may be synchronous or asynchronous)
  
• BPEL Process passing the Custom Header variable to another BPEL Process, invoked synchronously.
• BPEL Process passing the Custom Header variable to another BPEL Process, invoked asynchronously.

If you would like to send the parameters in the SOAP Header as per WS-Addressing specifications, you would need to modify the SOAP envelope and include these parameters, passed to the BPEL Process.

How to access the SOAP Headers in ESB or pass the SOAP Headers from ESB?

The steps for setting or passing the custom header in ESB is well documented over here. The only challenging piece of information that I could not find in the document was how to access the custom header variable when the ESB routing service is called directly using SOAP. You need to access the header variable as under,

<xsl:value-of select=
"ehdr:getRequestHeader('/soap:Header/ns2:
HeaderVariable/ns2:Data',
'soap=http://schemas.xmlsoap.org/soap/envelope/; ns2=http://xmlns.oracle.com/ESBProcess;')"/>

The function getRequestHeader() requires all the namespaces used in the Custom Header Variable along with soap namespace to be specified as the second parameter seperated by ';'

How to create and purge a AQ Queue/Topic

I was looking for testing the AQ Adapter and that forced me to look for steps on how to create and purge a AQ Queue/Topic? I referred the Article which explains all about AQ but in brief the steps are as under,

Creating a AQ Queue:

1. Grant the privileges to the Database User
2. Create a Queue/Topic table
3. Create a Queue/Topic
4. Start the Queue/Topic

In SQL terms,

1. Grant the privileges:

connect as sys;

grant connect, resource, aq_administrator_role to [USER] identified by [USER];
grant execute on sys.dbms_aqadm to [USER];
grant execute on sys.dbms_aq to [USER];
grant execute on sys.dbms_aqin to [USER];
grant execute on sys.dbms_aqjms to [USER];

2. Combining steps 2,3 and 4 - Create a Queue table, Queue and Start the Queue:

connect as [USER];
begin

DBMS_AQADM.CREATE_QUEUE_TABLE ( queue_table => 'XMLType_Payload_In', queue_payload_type => 'SYS.XMLType');
DBMS_AQADM.CREATE_QUEUE ( queue_name => 'XMLType_Payload_In', queue_table => 'XMLType_Payload_In');
DBMS_AQADM.START_QUEUE ( queue_name => 'XMLType_Payload_In');

end;
commit;

This will create a XMLType_Payload_In table to hold the messages of XMLType in the database.

Purging a AQ Queue:

DECLARE
po dbms_aqadm.aq$_purge_options_t;
BEGIN
po.block := FALSE;
DBMS_AQADM.PURGE_QUEUE_TABLE(
queue_table => 'XMLTYPE_PAYLOAD_IN',
purge_condition => NULL,
purge_options => po);
END;

The above block will purge all the records from the Queue.

Generating a Unique ID for every request in Oracle BPEL/ESB

A unique ID needs to be assigned to every asynchronous request for tracking the status/uniquely identifying the request message. The ID needs to be universally unique (UUID) that is generated by the Oracle SOA Systems (ESB/BPEL) to correlate the request for future reference. You can take advantage of the XPath Extension Function orcl:generate-guid() that comes out-of-the box and format it accordingly like this:

<copy>
<from expression="orcl:generate-guid()"/>
<to variable="uniqueId"/>
</copy>
<copy>
<from expression="orcl:format-string('{0}-{1}-{2}-{3}-{4}',
substring(bpws:getVariableData('uniqueId'),1,8),
substring(bpws:getVariableData('uniqueId'),9,4),
substring(bpws:getVariableData('uniqueId'),13,4), substring(bpws:getVariableData('uniqueId'),17,4), substring(bpws:getVariableData('uniqueId'),21, 12))"/>
<to variable="outputVariable" part="payload"
query="/client:LoanBrokerBPELProcessResponse/
client:LoanRequestId"/>
</copy>

This will create a variable called LoanRequestId with the required XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX format.

Error starting the SOA Suite

I tried starting the Oracle SOA Suite from the command prompt,
${SOA_HOME}opmn\bin>opmnctl.exe startall

But, I got an error as under,

opmnctl: starting opmn and all managed processes...
================================================================================
opmn id=localhost:6200
2 of 3 processes started.

ias-instanceid=oracleAS++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
ias-component/process-type/process-set:
default_group/oc4j_soa/default_group/

Error
--> Process (index=1,uid=936119581,pid=5296)
failed to start a managed process after the maximum retry limit
Log:
C:\SOA\product\10.1.3.1\OracleAS\opmn\logs\\default_group~oc4j_soa~default_group~1.log

After doing some research, I found that this is caused due to improper shutdown of the SOA suite. Some of the locks (JMS locks, DB locks ) are not released which needs to be manually removed from the file system.

Navigate to ${SOA_HOME}\j2ee\{OC4J Instance Name}\persistence and remove all the files.

Now start the SOA suite and it should work.

Issues with <exec> ANT task

The exec ant task is used to execute a system command. I came across the syntax that works on Windows but does not work on Linux.

Works on Windows:

<exec executable="${ant.home}/bin/ant" failonerror="true"
dir="${sandbox.path}/${processName}">
<arg value="-Dhttp.hostname"/>
<arg value="-Dhttp.hostname=${http.host}"/>
<arg value="-buildfile= build.xml"/>
</exec>

The tweak to make it working on the Linux,

<exec executable="${ant.home}/bin/ant" failonerror="true"
dir="${sandbox.path}/${processName}">
<arg value="-Dhttp.hostname"/>
<arg value="-Dhttp.hostname=${http.host}"/>
<arg value="-buildfile"/>
<arg value="build.xml"/>
</exec>