int numRunningJobs;
try {
numRunningJobs = jpaService.execute(new CoordJobGetRunningActionsCountJPAExecutor(jobId));
}
catch (JPAExecutorException je) {
throw new CommandException(je);
}
numActionsToStart = jobConcurrency - numRunningJobs;
if (numActionsToStart < 0) {
numActionsToStart = 0;
}
log.debug("concurrency=" + jobConcurrency + ", execution=" + jobExecution + ", numRunningJobs="
+ numRunningJobs + ", numLeftover=" + numActionsToStart);
// no actions to start
if (numActionsToStart == 0) {
log.warn("No actions to start! for jobId=" + jobId);
return null;
}
}
// get list of actions that are READY and fit in the concurrency and execution
List<CoordinatorActionBean> actions;
try {
actions = jpaService.execute(new CoordJobGetReadyActionsJPAExecutor(jobId, numActionsToStart, jobExecution));
}
catch (JPAExecutorException je) {
throw new CommandException(je);
}
log.debug("Number of READY actions = " + actions.size());
String user = coordJob.getUser();
String authToken = coordJob.getAuthToken();
// make sure auth token is not null
// log.denug("user=" + user + ", token=" + authToken);
int counter = 0;
for (CoordinatorActionBean action : actions) {
// continue if numActionsToStart is negative (no limit on number of
// actions), or if the counter is less than numActionsToStart
if ((numActionsToStart < 0) || (counter < numActionsToStart)) {
log.debug("Set status to SUBMITTED for id: " + action.getId());
// change state of action to SUBMITTED
action.setStatus(CoordinatorAction.Status.SUBMITTED);
// queue action to start action
queue(new CoordActionStartXCommand(action.getId(), user, authToken, action.getJobId()), 100);
try {
jpaService.execute(new org.apache.oozie.executor.jpa.CoordActionUpdateStatusJPAExecutor(action));
}
catch (JPAExecutorException je) {
throw new CommandException(je);
}
}
else {
break;
}