}
}
private void runBundleRecovery(){
XLog.Info.get().clear();
XLog log = XLog.getLog(getClass());
List<BundleActionBean> bactions = null;
try {
bactions = jpaService.execute(new BundleActionsGetWaitingOlderJPAExecutor(bundleOlderThan));
}
catch (JPAExecutorException ex) {
log.warn("Error reading bundle actions from database", ex);
return;
}
msg.append(", BUNDLE_ACTIONS : " + bactions.size());
for (BundleActionBean baction : bactions) {
try {
Services.get().get(InstrumentationService.class).get()
.incr(INSTRUMENTATION_GROUP, INSTR_RECOVERED_BUNDLE_ACTIONS_COUNTER, 1);
if (baction.getCoordId() == null) {
log.error("CoordId is null for Bundle action " + baction.getBundleActionId());
continue;
}
if (baction.getStatus() == Job.Status.PREP) {
BundleJobBean bundleJob = null;
if (jpaService != null) {
bundleJob = jpaService.execute(new BundleJobGetJPAExecutor(baction.getBundleId()));
}
if (bundleJob != null) {
Element bAppXml = XmlUtils.parseXml(bundleJob.getJobXml());
List<Element> coordElems = bAppXml.getChildren("coordinator", bAppXml.getNamespace());
for (Element coordElem : coordElems) {
Attribute name = coordElem.getAttribute("name");
if (name.getValue().equals(baction.getCoordName())) {
Configuration coordConf = mergeConfig(coordElem, bundleJob);
coordConf.set(OozieClient.BUNDLE_ID, baction.getBundleId());
queueCallable(new CoordSubmitXCommand(coordConf, bundleJob.getAuthToken(),
bundleJob.getId(), name.getValue()));
}
}
}
}
else if (baction.getStatus() == Job.Status.KILLED) {
queueCallable(new CoordKillXCommand(baction.getCoordId()));
}
else if (baction.getStatus() == Job.Status.SUSPENDED
|| baction.getStatus() == Job.Status.SUSPENDEDWITHERROR) {
queueCallable(new CoordSuspendXCommand(baction.getCoordId()));
}
else if (baction.getStatus() == Job.Status.RUNNING
|| baction.getStatus() == Job.Status.RUNNINGWITHERROR) {
queueCallable(new CoordResumeXCommand(baction.getCoordId()));
}
}
catch (Exception ex) {
log.error("Exception, {0}", ex.getMessage(), ex);
}
}
}