From: vasary.daniel Date: Mon, 8 Feb 2021 15:51:36 +0000 (+0000) Subject: git-tfs-id: [http://tfs.userrendszerhaz.hu:8080/tfs/DefaultCollection]$/MediaCube... X-Git-Url: http://git.useribm.hu/?a=commitdiff_plain;h=cd8a2c7fd59cdbbf68d4dbdaac0836fb24618d6c;p=mediacube.git git-tfs-id: [tfs.userrendszerhaz.hu:8080/tfs/DefaultCollection]$/MediaCube;C32135 --- diff --git a/server/user.jobengine.osgi.server/META-INF/MANIFEST.MF b/server/user.jobengine.osgi.server/META-INF/MANIFEST.MF index a427f2b0..fd6d94a5 100644 --- a/server/user.jobengine.osgi.server/META-INF/MANIFEST.MF +++ b/server/user.jobengine.osgi.server/META-INF/MANIFEST.MF @@ -3,7 +3,7 @@ Bundle-ManifestVersion: 2 Bundle-Name: Server Bundle-SymbolicName: user.jobengine.osgi.server;singleton:=true Bundle-Version: 1.0.0 -Service-Component: OSGI-INF/component.xml, OSGI-INF/componentBinder.xml +Service-Component: OSGI-INF/jobEngine.xml, OSGI-INF/jobEngineConfig.xml, OSGI-INF/componentBinder.xml Import-Package: com.fasterxml.jackson.databind;version="2.4.5", com.fasterxml.jackson.jaxrs.cfg;version="2.4.5", com.fasterxml.jackson.jaxrs.json;version="2.4.5", diff --git a/server/user.jobengine.osgi.server/OSGI-INF/componentBinder.xml b/server/user.jobengine.osgi.server/OSGI-INF/componentBinder.xml index 0b60807b..61bdea2e 100644 --- a/server/user.jobengine.osgi.server/OSGI-INF/componentBinder.xml +++ b/server/user.jobengine.osgi.server/OSGI-INF/componentBinder.xml @@ -2,5 +2,4 @@ - diff --git a/server/user.jobengine.osgi.server/OSGI-INF/component.xml b/server/user.jobengine.osgi.server/OSGI-INF/jobEngine.xml similarity index 61% rename from server/user.jobengine.osgi.server/OSGI-INF/component.xml rename to server/user.jobengine.osgi.server/OSGI-INF/jobEngine.xml index 02af0a3f..5feab105 100644 --- a/server/user.jobengine.osgi.server/OSGI-INF/component.xml +++ b/server/user.jobengine.osgi.server/OSGI-INF/jobEngine.xml @@ -1,11 +1,8 @@ - + diff --git a/server/user.jobengine.osgi.server/OSGI-INF/jobEngineConfig.xml b/server/user.jobengine.osgi.server/OSGI-INF/jobEngineConfig.xml new file mode 100644 index 00000000..2a258293 --- /dev/null +++ b/server/user.jobengine.osgi.server/OSGI-INF/jobEngineConfig.xml @@ -0,0 +1,8 @@ + + + + + + + + diff --git a/server/user.jobengine.osgi.server/src/user/jobengine/osgi/server/ComponentBinder.java b/server/user.jobengine.osgi.server/src/user/jobengine/osgi/server/ComponentBinder.java index 559a4228..0dd3f5ad 100644 --- a/server/user.jobengine.osgi.server/src/user/jobengine/osgi/server/ComponentBinder.java +++ b/server/user.jobengine.osgi.server/src/user/jobengine/osgi/server/ComponentBinder.java @@ -3,7 +3,6 @@ package user.jobengine.osgi.server; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; -import user.commons.configuration.IConfiguration; import user.commons.nexio.INexioAPI; import user.mediacube.metadata.interfaces.IMetadataProviderFactory; @@ -12,8 +11,6 @@ public class ComponentBinder { private static INexioAPI nexioAPI; private static IMetadataProviderFactory metadataProviderFactory; - private static IConfiguration systemConfig; - public static IMetadataProviderFactory getMetadataProviderFactory() { return metadataProviderFactory; } @@ -24,10 +21,6 @@ public class ComponentBinder { return nexioAPI; } - public static IConfiguration getSystemConfig() { - return systemConfig; - } - public synchronized void bindMetadataProviderFactory(Object service) { if (service instanceof IMetadataProviderFactory) { metadataProviderFactory = (IMetadataProviderFactory) service; @@ -42,13 +35,6 @@ public class ComponentBinder { } } - public synchronized void bindSystemConfiguration(Object service) { - if (service instanceof IConfiguration) { - systemConfig = (IConfiguration) service; - logger.info("IConfiguration service binded"); - } - } - public synchronized void unbindMetadataProviderFactory(Object service) { metadataProviderFactory = null; logger.info("IMetadataProviderFactory service unbinded"); diff --git a/server/user.jobengine.osgi.server/src/user/jobengine/server/IJobEngine.java b/server/user.jobengine.osgi.server/src/user/jobengine/server/IJobEngine.java index ae75af71..38bbd174 100644 --- a/server/user.jobengine.osgi.server/src/user/jobengine/server/IJobEngine.java +++ b/server/user.jobengine.osgi.server/src/user/jobengine/server/IJobEngine.java @@ -56,7 +56,7 @@ public interface IJobEngine { IJobRuntime getJobById(long jobId); - JobEngineConfiguration getJobEngineConfiguration(); + IJobEngineConfiguration getJobEngineConfiguration(); Map getJobs(); diff --git a/server/user.jobengine.osgi.server/src/user/jobengine/server/IJobEngineConfiguration.java b/server/user.jobengine.osgi.server/src/user/jobengine/server/IJobEngineConfiguration.java new file mode 100644 index 00000000..66b092ba --- /dev/null +++ b/server/user.jobengine.osgi.server/src/user/jobengine/server/IJobEngineConfiguration.java @@ -0,0 +1,28 @@ +package user.jobengine.server; + +import java.util.List; +import java.util.Map; + +import user.jobengine.server.ast.JobTemplate; +import user.jobengine.server.scheduler.ScheduledJob; +import user.jobengine.server.steps.IJobStep; + +public interface IJobEngineConfiguration { + + IJobStep createJobStep(String stepUnitName) throws Exception; + + Map getExecutors(); + + Map getPrograms(); + + Map getSchedules(); + + List getTemplates(); + + void load(IJobEngine jobEngine) throws Exception; + + void loadExecutors(IJobEngine jobEngine); + + void loadSchedules() throws Exception; + +} diff --git a/server/user.jobengine.osgi.server/src/user/jobengine/server/JobEngine.java b/server/user.jobengine.osgi.server/src/user/jobengine/server/JobEngine.java index c0862468..500b7447 100644 --- a/server/user.jobengine.osgi.server/src/user/jobengine/server/JobEngine.java +++ b/server/user.jobengine.osgi.server/src/user/jobengine/server/JobEngine.java @@ -21,15 +21,12 @@ import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; import com.ibm.nosql.json.api.BasicDBList; -import com.ibm.nosql.json.api.BasicDBObject; import user.commons.Job; import user.commons.JobStatus; import user.commons.cluster.ClusteredJob; -import user.commons.nosql.NoSQLUtils; import user.jobengine.db.IItemManager; import user.jobengine.db.ItemManagerData.SignalType; -import user.jobengine.osgi.server.ComponentBinder; import user.jobengine.server.actions.IStatusMachine; import user.jobengine.server.actions.JobAction; import user.jobengine.server.actions.StatusMachine; @@ -185,36 +182,40 @@ public class JobEngine implements IJobEngine { } private volatile boolean isRunning; + private volatile boolean isAllExecutionDisabled; private volatile boolean isScheduledExecutionDisabled; - private final BlockingQueue runQueue; private final BlockingQueue messageQueue; private final Map submittedJobs; private VM vm; private MessageDispatcher dispatcher; private JobChangedListenerChecker jobChangedListenerChecker; + private IUserMessageQueues userMessageQueues; + private final CyclicBarrier startUpBarrier; private final IStatusMachine statusMachine; private AtomicLong nextJobId; - private SchedulerService schedulerService = null; - private List jobChangedListenerList = new CopyOnWriteArrayList<>(); private Map remoteWorkers; private String masterServerAddress = System.getProperty("jobengine.master.server", ""); private final JobEngineRemote remoteEngine; + private ConcurrentHashMap keepAliveJobChangedListeners = new ConcurrentHashMap<>(); - private JobEngineConfiguration jobEngineConfiguration = new JobEngineConfiguration(); + + private IJobEngineConfiguration jobEngineConfiguration; /** * A MAM motor példányosítása. Ennek során létrejönnek a várakozási sorok, az ütemező szál, az üzenet kezelő szál. */ public JobEngine() { - isScheduledExecutionDisabled = ComponentBinder.getSystemConfig().value("jobs.scheduled-execution-disabled", false); + + // isScheduledExecutionDisabled = ComponentBinder.getSystemConfig().value("jobs.scheduled-execution-disabled", false); + isScheduledExecutionDisabled = true; runQueue = new PriorityBlockingQueue(); messageQueue = new LinkedBlockingQueue(); @@ -327,6 +328,13 @@ public class JobEngine implements IJobEngine { setItemManager(service); } + public synchronized void bindJobEngineConfiguration(Object service) { + if (service instanceof JobEngineConfiguration) { + jobEngineConfiguration = (IJobEngineConfiguration) service; + logger.info("IJobEngineConfiguration service binded"); + } + } + private void bootstrap() throws JobEngineException { //submit("fake-noparams.xml", "Bootstrap", null); } @@ -497,7 +505,7 @@ public class JobEngine implements IJobEngine { } @Override - public JobEngineConfiguration getJobEngineConfiguration() { + public IJobEngineConfiguration getJobEngineConfiguration() { return jobEngineConfiguration; } @@ -522,17 +530,11 @@ public class JobEngine implements IJobEngine { @Override public ScheduledJob getScheduledJob(String template) { - List jobs = NoSQLUtils.asList(schedulerService.getScheduleJobs()); - BasicDBObject importJob = null; - for (BasicDBObject job : jobs) { - if (template.equals(job.getString("template"))) { - importJob = job; - break; - } - } - if (importJob == null) + Map schedules = jobEngineConfiguration.getSchedules(); + ScheduledJob scheduledJob = schedules.get(template); + if (scheduledJob == null) throw new NullPointerException("Missing template: " + template); - return schedulerService.createScheduledJob(importJob); + return scheduledJob; } @Override @@ -709,8 +711,7 @@ public class JobEngine implements IJobEngine { if (schedulerService != null) schedulerService.shutdown(); - jobEngineConfiguration.loadTemplates(); - jobEngineConfiguration.loadExecutors(this); + jobEngineConfiguration.load(this); schedulerService = new SchedulerService(this); schedulerService.startup(); @@ -808,8 +809,7 @@ public class JobEngine implements IJobEngine { startUpBarrier.reset(); logger.info("JobEngine gracefully stopped"); - jobEngineConfiguration.loadTemplates(); - jobEngineConfiguration.loadExecutors(this); + jobEngineConfiguration.load(this); vm.start(); @@ -912,8 +912,7 @@ public class JobEngine implements IJobEngine { try { removeGarbage(); - jobEngineConfiguration.loadTemplates(); - jobEngineConfiguration.loadExecutors(this); + jobEngineConfiguration.load(this); vm = new VM(); dispatcher = new MessageDispatcher(); @@ -1052,4 +1051,5 @@ public class JobEngine implements IJobEngine { logger.error("Couldn't shutdown jobEngine", e); } } + } diff --git a/server/user.jobengine.osgi.server/src/user/jobengine/server/JobEngineConfiguration.java b/server/user.jobengine.osgi.server/src/user/jobengine/server/JobEngineConfiguration.java index 82729e8a..c4037986 100644 --- a/server/user.jobengine.osgi.server/src/user/jobengine/server/JobEngineConfiguration.java +++ b/server/user.jobengine.osgi.server/src/user/jobengine/server/JobEngineConfiguration.java @@ -17,18 +17,24 @@ import java.util.function.Consumer; import java.util.function.Predicate; import java.util.stream.Stream; +import org.apache.commons.lang.StringUtils; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; +import com.ibm.nosql.json.JSONUtil; +import com.ibm.nosql.json.api.BasicDBList; +import com.ibm.nosql.json.api.BasicDBObject; + import groovy.lang.GroovyClassLoader; import user.commons.configuration.IConfiguration; -import user.jobengine.osgi.server.ComponentBinder; +import user.commons.nosql.NoSQLUtils; import user.jobengine.server.ast.Encoder; import user.jobengine.server.ast.JobTemplate; import user.jobengine.server.ast.Parser; +import user.jobengine.server.scheduler.ScheduledJob; import user.jobengine.server.steps.IJobStep; -public class JobEngineConfiguration { +public class JobEngineConfiguration implements IJobEngineConfiguration { private static final Logger logger = LogManager.getLogger(); public static final String CONF_MEDIACUBE = "configuration/mediacube.json"; public static final String CONF_MAESTRO = "configuration/maestro.json"; @@ -38,19 +44,30 @@ public class JobEngineConfiguration { public static final String DIR_CLASSES = "jobs/classes"; public static final String DIR_TEMPLATES = "jobs/templates"; + private static IConfiguration systemConfig; private final Map executors = new LinkedHashMap(); private final Map programs = new LinkedHashMap(); + private final Map schedules = new LinkedHashMap(); private final List templates = Collections.synchronizedList(new ArrayList()); + private URLClassLoader stepsClassLoader = null; + public synchronized void bindSystemConfiguration(Object service) { + if (service instanceof IConfiguration) { + systemConfig = (IConfiguration) service; + logger.info("IConfiguration service binded"); + } + } + @SuppressWarnings("unchecked") + @Override public IJobStep createJobStep(String stepUnitName) throws Exception { IJobStep result = null; boolean isGroovyClass = stepUnitName.toLowerCase().endsWith(".java") || stepUnitName.toLowerCase().endsWith(".groovy"); if (stepsClassLoader == null) { - String stepClassesDir = ComponentBinder.getSystemConfig().getConfig(DIR_CLASSES); + String stepClassesDir = systemConfig.getConfig(DIR_CLASSES); Path path = Paths.get(stepClassesDir); URL[] urls = { path.toUri().toURL() }; stepsClassLoader = URLClassLoader.newInstance(urls, getClass().getClassLoader()); @@ -60,7 +77,7 @@ public class JobEngineConfiguration { if (isGroovyClass) { try (GroovyClassLoader gcl = new GroovyClassLoader(stepsClassLoader)) { - String stepsDir = ComponentBinder.getSystemConfig().getConfig(DIR_STEPS); + String stepsDir = systemConfig.getConfig(DIR_STEPS); stepClass = (Class) parseClassHierarchy(gcl, stepsDir, stepUnitName); } catch (Exception e) { throw e; @@ -74,22 +91,85 @@ public class JobEngineConfiguration { return result; } + private ScheduledJob createScheduledJob(BasicDBObject jobJSON) { + ScheduledJob sj = new ScheduledJob(); + String name = jobJSON.getString("name"); + String template = jobJSON.getString("template"); + sj.setJobId(String.format("%s.%s", template, name)); + if (StringUtils.isEmpty(name)) + sj.setJobName(template); + else + sj.setJobName(name); + + sj.setTemplateName(template); + sj.setCronExpressions(NoSQLUtils.asString(jobJSON, "cronexpression")); + sj.setExecuteOnStartup(NoSQLUtils.asBool(jobJSON, "executeimmediate")); + sj.setActive(NoSQLUtils.asBool(jobJSON, "active")); + + BasicDBList jobParams = (BasicDBList) jobJSON.get("parameters"); + if (jobParams != null) { + for (int p = 0; p < jobParams.size(); p++) { + BasicDBObject jsParam = (BasicDBObject) jobParams.get(p); + String n = jsParam.getString("name"); + Object v = jsParam.get("value"); + if (v instanceof BasicDBList) { + ArrayList list = (BasicDBList) v; + sj.setJobParameter(n, list); + } else { + if (v == null) + continue; + + String sv = String.valueOf(v); + + if ("null".equals(v)) + continue; + + Class clazz = null; + try { + clazz = Class.forName(jsParam.getString("type")); + Object value = clazz.getConstructor(new Class[] { String.class }).newInstance(sv); + sj.setJobParameter(n, value); + } catch (Exception e) { + logger.error(e); + } + } + + } + } + return sj; + } + + @Override public Map getExecutors() { return executors; } + @Override public Map getPrograms() { return programs; } + @Override + public Map getSchedules() { + return schedules; + } + + @Override public List getTemplates() { return templates; } + @Override + public void load(IJobEngine jobEngine) throws Exception { + loadTemplates(); + loadExecutors(jobEngine); + loadSchedules(); + } + + @Override public void loadExecutors(IJobEngine jobEngine) { InputStream stream = null; try { - IConfiguration systemConfig = ComponentBinder.getSystemConfig(); String configFilePath = systemConfig.getConfig(CONF_EXECUTORS); logger.info("Loading executors configuration {}", configFilePath); stream = Files.newInputStream(Paths.get(configFilePath)); @@ -128,6 +208,29 @@ public class JobEngineConfiguration { } } + @Override + public void loadSchedules() throws Exception { + getSchedules().clear(); + String configFilePath = systemConfig.getConfig(JobEngineConfiguration.CONF_SCHEDULES); + logger.info("Loading scheduler configuration file {}", configFilePath); + String jsonConfig = new String(Files.readAllBytes(Paths.get(configFilePath))); + BasicDBObject dbo = (BasicDBObject) JSONUtil.jsonToDbObject(jsonConfig); + BasicDBList scheduleJobs = (BasicDBList) dbo.get("joblist"); + for (int c = 0; c < scheduleJobs.size(); c++) { + BasicDBObject sjob = (BasicDBObject) scheduleJobs.get(c); + ScheduledJob sj = null; + try { + sj = createScheduledJob(sjob); + } catch (Exception e) { + logger.error("Error creating job: {}", sjob); + } + //skip inactive jobs + if (sj.isActive()) + getSchedules().put(sj.getTemplateName(), sj); + + } + } + private void loadTemplate(Path config) { InputStream stream = null; try { @@ -158,7 +261,6 @@ public class JobEngineConfiguration { public void loadTemplates() { programs.clear(); try { - IConfiguration systemConfig = ComponentBinder.getSystemConfig(); Path templatesPath = Paths.get(systemConfig.getConfig(DIR_TEMPLATES)); if (!templatesPath.toFile().isDirectory()) throw new FileNotFoundException(templatesPath + " is not directory!"); @@ -198,5 +300,4 @@ public class JobEngineConfiguration { result = cl.parseClass(path.toFile()); return result; } - } diff --git a/server/user.jobengine.osgi.server/src/user/jobengine/server/scheduler/SchedulerService.java b/server/user.jobengine.osgi.server/src/user/jobengine/server/scheduler/SchedulerService.java index 24b9231c..a30519b7 100644 --- a/server/user.jobengine.osgi.server/src/user/jobengine/server/scheduler/SchedulerService.java +++ b/server/user.jobengine.osgi.server/src/user/jobengine/server/scheduler/SchedulerService.java @@ -1,10 +1,8 @@ package user.jobengine.server.scheduler; -import java.nio.file.Files; -import java.nio.file.Paths; -import java.util.ArrayList; import java.util.Arrays; import java.util.List; +import java.util.Map; import java.util.UUID; import org.apache.commons.lang.StringUtils; @@ -28,15 +26,10 @@ import org.quartz.UnableToInterruptJobException; import org.quartz.impl.StdSchedulerFactory; import org.quartz.impl.matchers.KeyMatcher; -import com.ibm.nosql.json.JSONUtil; import com.ibm.nosql.json.api.BasicDBList; -import com.ibm.nosql.json.api.BasicDBObject; -import user.commons.configuration.IConfiguration; -import user.commons.nosql.NoSQLUtils; -import user.jobengine.osgi.server.ComponentBinder; import user.jobengine.server.IJobEngine; -import user.jobengine.server.JobEngineConfiguration; +import user.jobengine.server.IJobEngineConfiguration; /** * Jobok schedulalt inditasa. @@ -47,10 +40,6 @@ public class SchedulerService { private static final Logger logger = LogManager.getLogger(); private static final String CRON_EXPRESSIONS_DELIMITER = ";"; - public static T parseObjectFromString(String s, Class clazz) throws Exception { - return clazz.getConstructor(new Class[] { String.class }).newInstance(s); - } - private Scheduler scheduler = null; private IJobEngine jobEngine = null; private BasicDBList scheduleJobs; @@ -61,56 +50,6 @@ public class SchedulerService { this.jobEngine = jobEngine; } - public ScheduledJob createScheduledJob(BasicDBObject jobJSON) { - ScheduledJob sj = new ScheduledJob(); - // sj.setJobId(String.valueOf(++schedulerJobId)); - String name = jobJSON.getString("name"); - String template = jobJSON.getString("template"); - sj.setJobId(String.format("%s.%s", template, name)); - if (StringUtils.isEmpty(name)) - sj.setJobName(template); - else - sj.setJobName(name); - - sj.setTemplateName(template); - sj.setCronExpressions(NoSQLUtils.asString(jobJSON, "cronexpression")); - sj.setExecuteOnStartup(NoSQLUtils.asBool(jobJSON, "executeimmediate")); - sj.setActive(NoSQLUtils.asBool(jobJSON, "active")); - - BasicDBList jobParams = (BasicDBList) jobJSON.get("parameters"); - if (jobParams != null) { - for (int p = 0; p < jobParams.size(); p++) { - BasicDBObject jsParam = (BasicDBObject) jobParams.get(p); - String n = jsParam.getString("name"); - Object v = jsParam.get("value"); - if (v instanceof BasicDBList) { - ArrayList list = (BasicDBList) v; - sj.setJobParameter(n, list); - } else { - if (v == null) - continue; - - String sv = String.valueOf(v); - - if ("null".equals(v)) - continue; - - Class clazz = null; - try { - clazz = Class.forName(jsParam.getString("type")); - Object value = parseObjectFromString(sv, clazz); - sj.setJobParameter(n, value); - } catch (Exception e) { - logger.error(e); - } - } - - } - } - - return sj; - } - /** * Regisztralt jobok azonnali futtatasa. * @@ -172,33 +111,6 @@ public class SchedulerService { } } - /** - * Elore konfiguralt, idozitett (serice) jobok inditasa - * - * @throws Exception - */ - private void loadStartupJobsFromConfig() throws Exception { - IConfiguration systemConfig = ComponentBinder.getSystemConfig(); - String configFilePath = systemConfig.getConfig(JobEngineConfiguration.CONF_SCHEDULES); - logger.info("Loading scheduler configuration file {}", configFilePath); - String jsonConfig = new String(Files.readAllBytes(Paths.get(configFilePath))); - BasicDBObject dbo = (BasicDBObject) JSONUtil.jsonToDbObject(jsonConfig); - scheduleJobs = (BasicDBList) dbo.get("joblist"); - for (int c = 0; c < scheduleJobs.size(); c++) { - BasicDBObject sjob = (BasicDBObject) scheduleJobs.get(c); - ScheduledJob sj = null; - try { - sj = createScheduledJob(sjob); - } catch (Exception e) { - logger.error("Error creating job: {}", sjob); - } - //skip inactive jobs - if (!sj.isActive()) - continue; - schedule(sj); - } - } - public void register(ScheduledJob job) throws SchedulerException { String jobID = job.getJobId(); JobDataMap jobDataMap = new JobDataMap(); @@ -277,7 +189,10 @@ public class SchedulerService { public void startup() throws Exception { scheduler = StdSchedulerFactory.getDefaultScheduler(); scheduler.start(); - loadStartupJobsFromConfig(); + IJobEngineConfiguration jobEngineConfiguration = jobEngine.getJobEngineConfiguration(); + Map schedules = jobEngineConfiguration.getSchedules(); + for (ScheduledJob job : schedules.values()) + schedule(job); } public void unregisterJob(ScheduledJob job) throws SchedulerException { diff --git a/server/user.mediacube.gui/src/user/jobengine/zk/model/JobSelectorModel.java b/server/user.mediacube.gui/src/user/jobengine/zk/model/JobSelectorModel.java index 482fe685..45c69860 100644 --- a/server/user.mediacube.gui/src/user/jobengine/zk/model/JobSelectorModel.java +++ b/server/user.mediacube.gui/src/user/jobengine/zk/model/JobSelectorModel.java @@ -53,8 +53,8 @@ public class JobSelectorModel extends BaseModel { public void execute() { if (selectedJob == null) return; - SchedulerService scheduler = jobEngine.getScheduler(); - ScheduledJob scheduledJob = scheduler.createScheduledJob(selectedJob); + + ScheduledJob scheduledJob = jobEngine.getScheduledJob(""); scheduledJob.setJobEngine(jobEngine); try { scheduledJob.doManualJob();