public static JobRecord registerNewJobRecord(UpdateSpec updateSpec, JobSetting[] settings,
JobRecord generatorJob, String graphGUID, Job<?> jobInstance, Object[] params) {
Key rootKey = (null == generatorJob ? null : generatorJob.getRootJobKey());
Key generatorKey = (null == generatorJob ? null : generatorJob.getKey());
JobRecord jobRecord = new JobRecord(rootKey, generatorKey, graphGUID, jobInstance, settings);
updateSpec.setRootJobKey(jobRecord.getRootJobKey());
// Add slots to the RunBarrier corresponding to the input parameters
for (Object param : params) {
Value<?> value;
if (null != param && param instanceof Value<?>) {
value = (Value<?>) param;
} else {
value = new ImmediateValue<Object>(param);
}
registerSlotsWithBarrier(updateSpec, value, jobRecord.getRootJobKey(), generatorKey,
graphGUID, jobRecord.getRunBarrierInflated());
}
if (0 == jobRecord.getRunBarrierInflated().getWaitingOnKeys().size()) {
// If the run barrier is not waiting on anything, add a phantom filled
// slot in order to trigger a HandleSlotFilledTask in order to trigger
// a RunJobTask.
Slot slot = new Slot(jobRecord.getRootJobKey(), generatorKey, graphGUID);
jobRecord.getRunBarrierInflated().addPhantomArgumentSlot(slot);
registerSlotFilled(updateSpec, slot, null);
}
// Register the newly created objects with the UpdateSpec.
// The slots in the run Barrier have already been registered
// and the finalize Barrier doesn't have any slots yet.
// Any HandleSlotFilledTasks have also been registered already.
Group updateGroup = updateSpec.getNonTransactionalGroup();
updateGroup.includeBarrier(jobRecord.getRunBarrierInflated());
updateGroup.includeBarrier(jobRecord.getFinalizeBarrierInflated());
updateGroup.includeSlot(jobRecord.getOutputSlotInflated());
updateGroup.includeJob(jobRecord);
updateGroup.includeJobInstanceRecord(jobRecord.getJobInstanceInflated());
return jobRecord;
}