Updated Branches: refs/heads/ode-1.3.6.x 6a6b9ca77 -> 8cb723667
ODE-942: fixing a scheduler bug. Thanks Waruna for the patch. Project: http://git-wip-us.apache.org/repos/asf/ode/repo Commit: http://git-wip-us.apache.org/repos/asf/ode/commit/8cb72366 Tree: http://git-wip-us.apache.org/repos/asf/ode/tree/8cb72366 Diff: http://git-wip-us.apache.org/repos/asf/ode/diff/8cb72366 Branch: refs/heads/ode-1.3.6.x Commit: 8cb723667d4221cef2bcbe818d03c635294132a1 Parents: 6a6b9ca Author: Tammo van Lessen <[email protected]> Authored: Sun Sep 15 22:22:42 2013 +0200 Committer: Tammo van Lessen <[email protected]> Committed: Sun Sep 15 22:22:42 2013 +0200 ---------------------------------------------------------------------- .../ode/scheduler/simple/SimpleScheduler.java | 45 ++++++++++++++++++-- 1 file changed, 41 insertions(+), 4 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/ode/blob/8cb72366/scheduler-simple/src/main/java/org/apache/ode/scheduler/simple/SimpleScheduler.java ---------------------------------------------------------------------- diff --git a/scheduler-simple/src/main/java/org/apache/ode/scheduler/simple/SimpleScheduler.java b/scheduler-simple/src/main/java/org/apache/ode/scheduler/simple/SimpleScheduler.java index 7438f9f..7b7c7c2 100644 --- a/scheduler-simple/src/main/java/org/apache/ode/scheduler/simple/SimpleScheduler.java +++ b/scheduler-simple/src/main/java/org/apache/ode/scheduler/simple/SimpleScheduler.java @@ -379,12 +379,12 @@ public class SimpleScheduler implements Scheduler, TaskRunner { boolean nearfuture = !immediate && when.getTime() <= ctime + _nearFutureInterval; try { if (immediate) { - // Immediate scheduling means we put it in the DB for safe keeping - _db.insertJob(job, _nodeId, true); - + // Immediate scheduling means we put it in the DB for safe keeping. // And add it to our todo list . if (_outstandingJobs.size() < _todoLimit) { - addTodoOnCommit(job); + saveAndAddTodoOnCommit(job, _nodeId, true); + } else { + saveTodoOnCommit(job, _nodeId, true); } __log.debug("scheduled immediate job: " + job.jobId); } else if (nearfuture) { @@ -621,6 +621,43 @@ public class SimpleScheduler implements Scheduler, TaskRunner { _exec.submit(new RunJob(job, _polledRunnableProcessor)); } + private void saveAndAddTodoOnCommit(final Job job, final String nodeId, final boolean loaded) { + registerSynchronizer(new Synchronizer() { + public void afterCompletion(boolean success) { + if (success) { + try { + _db.insertJob(job, nodeId, loaded); + } catch (DatabaseException dbe) { + __log.error("Could not save job on commit. Will add it to in-mem queue anyway.", dbe); + throw new ContextException(dbe.getMessage(), dbe); + } + enqueue(job); + } + } + + public void beforeCompletion() { + } + }); + } + + private void saveTodoOnCommit(final Job job, final String nodeId, final boolean loaded) { + registerSynchronizer(new Synchronizer() { + public void afterCompletion(boolean success) { + if (success) { + try { + _db.insertJob(job, nodeId, loaded); + } catch (DatabaseException dbe) { + __log.error("Could not save job on commit.", dbe); + throw new ContextException(dbe.getMessage(), dbe); + } + } + } + + public void beforeCompletion() { + } + }); + } + private void addTodoOnCommit(final Job job) { registerSynchronizer(new Synchronizer() { public void afterCompletion(boolean success) {
