if (load_ex != null) {
// some messages could not be loaded
LoadException processing = load_ex;
while (processing != null) {
TransactionUID tid = (TransactionUID)processing.getKey();
TransactionInfo ti = (TransactionInfo)processing.getValue();
if (tid == null && ti == null) {
logger.log(Logger.WARNING, BrokerResources.E_INTERNAL_ERROR,
"Both key and value for a transactions entry"
+ " are corrupted with key exception "+
processing.getKeyCause().getMessage()+
" and value exception "+ processing.getValueCause());
processing = processing.getNextException();
continue;
}
if (tid == null ) {
// at this point, there is nothing we can do ...
// store with valid key
// improve when we address 5060661
logger.logStack(Logger.WARNING, BrokerResources.W_TRANS_ID_CORRUPT,
ti.toString(), processing.getKeyCause());
processing = processing.getNextException();
continue;
}
if (ti == null) {
// if we dont know ... so make it prepared
logger.log(Logger.WARNING, Globals.getBrokerResources().getKString(
BrokerResources.W_TRAN_INFO_CORRUPTED, tid.toString()));
TransactionState ts = new TransactionState(
AutoRollbackType.NOT_PREPARED, 0, true);
ts.setState(TransactionState.PREPARED);
try {
store.storeTransaction(tid, ts, false);
} catch (Exception ex) {
logger.logStack(Logger.WARNING, BrokerResources.E_INTERNAL_ERROR,
"Error updating transaction " + tid, ex);
}
processing = processing.getNextException();
continue;
}
if (ti.getType() == TransactionInfo.TXN_NOFLAG) {
logger.logStack(Logger.WARNING, Globals.getBrokerResources().getKString(
BrokerResources.W_TXN_TYPE_CORRUPTED, tid+"["+ti.toString()+"]",
TransactionInfo.toString(TransactionInfo.TXN_LOCAL)),
processing.getValueCause());
TransactionState ts = new TransactionState(
AutoRollbackType.NOT_PREPARED, 0, true);
ts.setState(TransactionState.PREPARED);
try {
store.storeTransaction(tid, ts, false);
} catch (Exception ex) {
logger.logStack(Logger.WARNING, BrokerResources.E_INTERNAL_ERROR,
"Error updating transaction " + tid, ex);
}
processing = processing.getNextException();
continue;
}
if (ti.getTransactionState() == null) {
logger.log(Logger.WARNING, BrokerResources.W_TRANS_STATE_CORRUPT,
tid, processing.getValueCause());
TransactionState ts = new TransactionState(
AutoRollbackType.NOT_PREPARED, 0, true);
ts.setState(TransactionState.PREPARED);
try {
store.storeTransaction(tid, ts, false);
} catch (Exception ex) {
logger.logStack(Logger.WARNING, BrokerResources.E_INTERNAL_ERROR,
"Error updating transaction " + tid, ex);
}
processing = processing.getNextException();
continue;
}
if (ti.getType() == TransactionInfo.TXN_CLUSTER && ti.getTransactionBrokers() == null) {
logger.log(Logger.WARNING, BrokerResources.W_CLUSTER_TXN_BROKER_INFO_CORRUPTED,
tid, processing.getValueCause());
store.storeClusterTransaction(tid, ti.getTransactionState(), null, false);
processing = processing.getNextException();
continue;
}
if (ti.getType() == TransactionInfo.TXN_REMOTE &&
ti.getTransactionHomeBroker() == null) {
logger.log(Logger.WARNING, BrokerResources.W_REMOTE_TXN_HOME_BROKER_INFO_CORRUPTED,
tid, processing.getValueCause());
store.storeRemoteTransaction(tid, ti.getTransactionState(), null, null, false);
processing = processing.getNextException();
continue;
}
logger.log(Logger.ERROR, "XXXI18N Internal Error: unknown load error for TUID="
+tid+"["+ti.toString()+"]");
} // end while
}
// now look at acks load exception
load_ex = Globals.getStore().getLoadTransactionAckException();
if (load_ex != null) {
// some messages could not be loaded
LoadException processing = load_ex;
while (processing != null) {
TransactionUID tid = (TransactionUID)processing.getKey();
TransactionAcknowledgement ta[] =
(TransactionAcknowledgement[])processing.getValue();
if (tid == null && ta == null) {
logger.log(Logger.WARNING,
BrokerResources.E_INTERNAL_ERROR,
"both key and value for a Transaction Ack entry"
+ " are corrupted");
processing = processing.getNextException();
continue;
}
if (tid == null ) {
// at this point, there is nothing we can do ...
// store with valid key
// improve when we address 5060661
logger.log(Logger.WARNING,
BrokerResources.W_TRANS_ID_CORRUPT,
ta.toString());
processing = processing.getNextException();
continue;
}
//ta == null, nothing we can do, remove it
logger.log(Logger.WARNING,
BrokerResources.W_TRANS_ACK_CORRUPT,
tid.toString());
try {
store.removeTransactionAck(tid, false);
} catch (Exception ex) {
logger.logStack(Logger.WARNING, BrokerResources.E_INTERNAL_ERROR,
"Error updating transaction acks " + tid, ex);
}
} // end while
}
logger.log(Logger.INFO, BrokerResources.I_PROCESSING_TRANS);
// OK -> first load the list of pending
// transactions
HashMap trans = store.getAllTransactionStates();
// Write some info about the transactions to the log file
// for informational purposes.
logTransactionInfo(trans, AUTO_ROLLBACK);
// list of transactions which need to be cleaned up
HashSet clearTrans = new HashSet(trans.size());
HashSet clearAckOnlyTrans = new HashSet(trans.size());
HashMap openTransactions = new HashMap();
HashMap inprocessAcks = new HashMap();
HashMap committingTransactions = new HashMap();
// loop through the list of transactions
// placing each on the various lists
int totalN = 0, clusterN = 0, remoteN = 0;
int prepareN = 0, commitN = 0, rollbackN = 0;
int prepareCN = 0, commitCN = 0, commitWaitCN = 0;
Iterator itr = trans.entrySet().iterator();
while (itr.hasNext()) {
totalN++;
try {
Map.Entry entry = (Map.Entry)itr.next();
TransactionUID tid = (TransactionUID)entry.getKey();
TransactionInfo tif = (TransactionInfo)entry.getValue();
TransactionState ts = tif.getTransactionState();
TransactionAcknowledgement ta[] = store.getTransactionAcks(tid);
int state = ts.getState();
if (DEBUG) {
logger.log(Logger.INFO, "Load transaction: TUID="+tid+
"["+TransactionState.toString(state)+
(ts.getCreationTime() == 0 ? "":" createTime="+ts.getCreationTime())+"]");
}
switch (state) {
// no longer valid, ignore
case TransactionState.CREATED:
clearTrans.add(tid);
break;
case TransactionState.PREPARED:
prepareN++;
// if autorollback, fall through to rollback
if (!AUTO_ROLLBACK) {
// nothing happens w/ preparedTransactions
// they go back into the list
// We don't persist this because it is already
// in the store
addTransactionID(tid, ts, false);
if (tif.getType() == TransactionInfo.TXN_CLUSTER) {
logClusterTransaction(tid, ts,
tif.getTransactionBrokers(), true, false);
prepareCN++;
}
openTransactions.put(tid, Boolean.TRUE);
// put messages in the orphan list
break;
}
// rollback -> we didnt complete
case TransactionState.STARTED:
case TransactionState.COMPLETE:
case TransactionState.ROLLEDBACK:
case TransactionState.FAILED:
case TransactionState.INCOMPLETE:
addTransactionID(tid, ts, false);
if (tif.getType() == TransactionInfo.TXN_CLUSTER) {
logClusterTransaction(tid, ts,
tif.getTransactionBrokers(), true, false);
}
openTransactions.put(tid, Boolean.FALSE);
clearTrans.add(tid);
ts.setState(TransactionState.ROLLEDBACK);
rollbackN++;
if (state == TransactionState.PREPARED) {
clearAckOnlyTrans.add(tid);
try {
updateState(tid, TransactionState.ROLLEDBACK, true);
} catch (Exception e) {
logger.logStack(Logger.WARNING,
"Unable to update auto-rollback PREPARED transaction "+tid+" state to ROLLEDBACK", e);
}
}
break;
case TransactionState.COMMITTED:
commitN++;
committingTransactions.put(tid, "");
if (tif.getType() == TransactionInfo.TXN_CLUSTER) {
commitCN++;
boolean completed = true;
TransactionBroker[] brokers = tif.getTransactionBrokers();
logClusterTransaction(tid, ts, brokers, false, false);
for (int i = 0; brokers != null && i < brokers.length; i++) {
completed = brokers[i].isCompleted();
if (!completed) {
if (DEBUG_CLUSTER_TXN) {
logger.log(logger.INFO, "COMMITTED cluster transaction "+tid+", incomplete broker:"+ brokers[i]);
}
break;
}
}
if (!completed) commitWaitCN++;
} else {
addTransactionID(tid, ts, false);
}
clearTrans.add(tid);
break;
default: logger.log(logger.ERROR,
"Internal Error unexpected transaction state:"+
TransactionState.toString(state)+ " TUID="+tid+", set to PREPARE");
addTransactionID(tid, ts, false);
if (tif.getType() == TransactionInfo.TXN_CLUSTER) {
logClusterTransaction(tid, ts,
tif.getTransactionBrokers(), true, false);
}
updateState(tid, TransactionState.PREPARED, true);
openTransactions.put(tid, Boolean.TRUE);
}