* without committing, thus doing a rollback.
*/
@SuppressWarnings({"ThrowFromFinallyBlock", "unchecked"})
public final T call() throws CommandException {
XLog.Info.get().setParameters(logInfo);
XLog log = XLog.getLog(getClass());
log.trace(logMask, "Start");
Instrumentation.Cron cron = new Instrumentation.Cron();
cron.start();
callables = new ArrayList<XCallable<Void>>();
delayedCallables = new ArrayList<XCallable<Void>>();
exceptionCallables = new ArrayList<XCallable<Void>>();
delay = 0;
S store = null;
boolean exception = false;
try {
if (withStore) {
store = (S) Services.get().get(StoreService.class).getStore(getStoreClass());
store.beginTrx();
}
T result = execute(store);
/*
*
* if (store != null && log != null) { log.info(XLog.STD,
* "connection log from store Flush Mode {0} ",
* store.getFlushMode()); }
*/
if (withStore) {
if (store == null) {
throw new IllegalStateException("WorkflowStore should not be null");
}
if (FaultInjection.isActive("org.apache.oozie.command.SkipCommitFaultInjection")) {
throw new RuntimeException("Skipping Commit for Failover Testing");
}
store.commitTrx();
}
// TODO figure out the reject due to concurrency problems and remove
// the delayed queuing for callables.
boolean ret = Services.get().get(CallableQueueService.class).queueSerial(callables, 10);
if (ret == false) {
logQueueCallableFalse(callables);
}
ret = Services.get().get(CallableQueueService.class).queueSerial(delayedCallables, delay);
if (ret == false) {
logQueueCallableFalse(delayedCallables);
}
return result;
}
catch (XException ex) {
log.error(logMask | XLog.OPS, "XException, {0}", ex.getMessage(), ex);
if (store != null) {
log.info(XLog.STD, "XException - connection logs from store {0}, {1}", store.getConnection(), store
.isClosed());
}
exception = true;
if (store != null && store.isActive()) {
try {
store.rollbackTrx();
}
catch (RuntimeException rex) {
log.warn(logMask | XLog.OPS, "openjpa error, {0}, {1}", name, rex.getMessage(), rex);
}
}
// TODO figure out the reject due to concurrency problems and remove
// the delayed queuing for callables.
boolean ret = Services.get().get(CallableQueueService.class).queueSerial(exceptionCallables, 10);
if (ret == false) {
logQueueCallableFalse(exceptionCallables);
}
if (ex instanceof CommandException) {
throw (CommandException) ex;
}
else {
throw new CommandException(ex);
}
}
catch (Exception ex) {
log.error(logMask | XLog.OPS, "Exception, {0}", ex.getMessage(), ex);
exception = true;
if (store != null && store.isActive()) {
try {
store.rollbackTrx();
}
catch (RuntimeException rex) {
log.warn(logMask | XLog.OPS, "openjpa error, {0}, {1}", name, rex.getMessage(), rex);
}
}
throw new CommandException(ErrorCode.E0607, name, ex.getMessage(), ex);
}
catch (Error er) {
log.error(logMask | XLog.OPS, "Error, {0}", er.getMessage(), er);
exception = true;
if (store != null && store.isActive()) {
try {
store.rollbackTrx();
}
catch (RuntimeException rex) {
log.warn(logMask | XLog.OPS, "openjpa error, {0}, {1}", name, rex.getMessage(), rex);
}
}
throw er;
}
finally {
FaultInjection.deactivate("org.apache.oozie.command.SkipCommitFaultInjection");
cron.stop();
instrumentation.addCron(INSTRUMENTATION_GROUP, name, cron);
incrCommandCounter(1);
log.trace(logMask, "End");
if (locks != null) {
for (LockToken lock : locks) {
lock.release();
}
locks.clear();
}
if (store != null) {
if (!store.isActive()) {
try {
store.closeTrx();
}
catch (RuntimeException rex) {
if (exception) {
log.warn(logMask | XLog.OPS, "openjpa error, {0}, {1}", name, rex.getMessage(), rex);
}
else {
throw rex;
}
}
}
else {
log.warn(logMask | XLog.OPS, "transaction is not committed or rolled back before closing entitymanager.");
}
}
}
}