Automatic recovery program for pending BPEL call back messages

BPEL engine maintains all async call back messages into database table called dlv_message. You can see such all messages in BPEL console call-back manual recovery area.The query being used by bpel console is joined on dlv_message and work_item tables.This query simply picks up all call back messages which are undelivered and have not been modified with in certain threshold time.

Call-back messages are processed in following steps

• BPEL engine assigns the call-back message to delivery service
• Delivery service saves the message into dlv_message table with state 'UNDELIVERED-0'
• Delivery service schedules a dispatcher thread to process message asynchronously
• Dispatcher thread enqueues message into JMS queue
• Message is picked up by MDB
• MDB delivers the message to actual BPEL process  waiting for call-back and changes state to 'HANDLED=2'

So given above steps, there is always possibility that message is available in dlv_message table but MDB is failed in delivering it to BPEL process which keeps message always in state= 0.

Following program can be tailored to suite one's own requirements to recover from such state-0 messages.

Note:- This program contains logic to recover from invocation and call-back messages. Please comment out appropriately.

package bpelrecovery;
import com.oracle.bpel.client.*;
import com.oracle.bpel.client.util.SQLDefs;
import com.oracle.bpel.client.util.WhereCondition;
import java.util.ArrayList;
import java.util.Hashtable;
import java.util.List;
import javax.naming.Context;

public class bpelrecovery {

public bpelrecovery() {

}

public static void main(String[] args) {

bpelrecovery recover = new bpelrecovery();
String rtc = "";
try{
rtc = recover.doRecover();
}
catch (Exception e){
e.printStackTrace();
}
}
rivate void recoverCallbackMessages(List messages)
throws Exception
{

String messageGuids[] = new String[messages.size()];
for(int i = 0; i < messages.size(); i++)
{
ICallbackMetaData callbackMetadata = (ICallbackMetaData)messages.get(i);
String messageGuid = callbackMetadata.getMessageGUID();

messageGuids[i] = messageGuid;
System.err.println((new StringBuilder()).append("recovering callback message =
").append(messageGuids[i]).append(" process
[").append(callbackMetadata.getProcessId()).append("(").append(callbackMetadata.getRevisionTag()).ap
pend(")] domain [").append(callbackMetadata.getDomainId()).append("]").toString());

}

Locator locator = getLocator();
IBPELDomainHandle domainHandle = locator.lookupDomain();
domainHandle.recoverCallbackMessages(messageGuids);

}

public String doRecover() throws Exception{
// Connect to domain "default"

try{

System.out.println("doRecover() instantiating locator...");

Locator locator = getLocator();

System.out.println("doRecover() instantiated locator for domain " +

locator.lookupDomain().getDomainId());

// look for Invoke messages in need of recovery

StringBuffer buf1 = new StringBuffer();

WhereCondition where = new WhereCondition(buf1.append(SQLDefs.IM_state).append( " = "

).append(IDeliveryConstants.STATE_UNRESOLVED ).toString() );

System.out.println("doRecover() instantiating IInvokeMetaData... with where = "+ where.getClause());

IInvokeMetaData imd1[] = locator.listInvokeMessages(where);

System.out.println("doRecover() instantiated IInvokeMetaData");

// iterate thru the list
List l1 = new ArrayList();

for (Object o:imd1){

l1.add(o);

}

// See how many INVOKES are in the recovery zone

System.out.println("doRecover() instantiated IInvokeMetaData size = " +l1.size());

// look for Callback messages in need of recovery

StringBuffer buf = new StringBuffer();

where = new WhereCondition(buf.append(SQLDefs.DM_state).append( " = "

).append(IDeliveryConstants.TYPE_callback_soap ).toString() );

System.out.println("doRecover() instantiating ICallbackMetaData... with where = "+

where.getClause());

ICallbackMetaData imd[] = locator.listCallbackMessages(where);

System.out.println("doRecover() instantiated ICallbackMetaData");
//

// recover

//

List l = new ArrayList();

for (Object o:imd){

l.add(o);

}

 recoverCallbackMessages(l);

}

catch (Exception e){

e.printStackTrace();

}
return "done";

}

public Locator getLocator(){

System.out.println("getLocator() start");

Locator locator = null;
// set JNDI properties for BPEL lookup

String jndiProviderUrl = "opmn:ormi://localhost:6003:oc4j_soa/orabpel";

String jndiFactory = "com.evermind.server.rmi.RMIInitialContextFactory";

String jndiUsername = "oc4jadmin";

String jndiPassword = "welcome1";
Hashtable jndi = new Hashtable();

jndi.put(Context.PROVIDER_URL, jndiProviderUrl);

jndi.put(Context.INITIAL_CONTEXT_FACTORY, jndiFactory);

jndi.put(Context.SECURITY_PRINCIPAL, jndiUsername);

jndi.put(Context.SECURITY_CREDENTIALS, jndiPassword);

jndi.put("dedicated.connection", "true");

try{

System.out.println("getLocator() instantiating locator...");

locator = new Locator("default", "welcome1", jndi);

System.out.println("getLocator() instantiated locator");

}

catch (Exception e){

System.out.println("getLocator() error");

e.printStackTrace();

}

return locator;

}
}