JobExecutor enhencement proposition to allow 3rd framework i

[p.pasqual]

Hi all,

I work currently on integartion of Jbpm solution under our Software which is base on Spring Framwork : using Spring bean for configuration and AOP interception for transactional and logging stuff.

An Custom ObjectFactory implementation permit to use bean configuration
made under "Spring Beans" if existe and use default ObjectFactoryImpl if not.

JobExecutor 's Fields are all "package" , and there is no relative "public" setter, soo it's impossible to set values from a Spring Bean Context. One of the simplest way to solve this is to simply add setters for all field present under jbpm.cfg.xml file.

Un other point is About posibility to use Spring AOP transaction interception
during assynchronous Job Execution (ie Using transaction context of hosting platform of course if provided). For this i propose to un-couple JbpmContext technical (aquire/close/exception handling) and functional (Job Execution) stuff using Service/Callback writing pattern.

Enhencement Resume :
1/ JobExecutor setter Add
2/ JobExecutorThread execution actions under host provided transactional context.

To give you an idee i send you a "draft" patch corresponding to my implementation. I made it to limite impact at the minimum on code base. At the first aproche i only refactor JobExecutorThread.executeJob() to use Callback but i think it's possible to extends this methodology to all other point where there is JbpmContext creation.

i waiting for your feed back

best regards

Index: src/jpdl/org/jbpm/job/executor/JobExecutor.java
===================================================================
--- src/jpdl/org/jbpm/job/executor/JobExecutor.java (revision 20)
+++ src/jpdl/org/jbpm/job/executor/JobExecutor.java (working copy)
@@ -15,11 +15,14 @@
 import org.apache.commons.logging.LogFactory;
 import org.jbpm.JbpmConfiguration;

+
 public class JobExecutor implements Serializable {

 private static final long serialVersionUID = 1L;

 JbpmConfiguration jbpmConfiguration;
+ JobExecutorService jobExecutorService;
+
 String name;
 int nbrOfThreads;
 int idleInterval;
@@ -164,6 +167,57 @@
 public int getNbrOfThreads() {
 return nbrOfThreads;
 }
+
+ public void setLockMonitorThread(LockMonitorThread in_lockMonitorThread) {
+ lockMonitorThread = in_lockMonitorThread;
+ }
+
+ public void setJbpmConfiguration(JbpmConfiguration in_jbpmConfiguration) {
+ jbpmConfiguration = in_jbpmConfiguration;
+ }
+
+ public void setName(String in_name) {
+ name = in_name;
+ }
+
+ public void setIdleInterval(int in_idleInterval) {
+ idleInterval = in_idleInterval;
+ }
+
+ public void setMaxIdleInterval(int in_maxIdleInterval) {
+ maxIdleInterval = in_maxIdleInterval;
+ }
+
+ public void setHistoryMaxSize(int in_historyMaxSize) {
+ historyMaxSize = in_historyMaxSize;
+ }
+
+ public void setMaxLockTime(int in_maxLockTime) {
+ maxLockTime = in_maxLockTime;
+ }
+
+ public void setLockMonitorInterval(int in_lockMonitorInterval) {
+ lockMonitorInterval = in_lockMonitorInterval;
+ }
+
+ public void setLockBufferTime(int in_lockBufferTime) {
+ lockBufferTime = in_lockBufferTime;
+ }
+
+ public void setNbrOfThreads(int in_nbrOfThreads) {
+ nbrOfThreads = in_nbrOfThreads;
+ }
+
+ public JobExecutorService getJobExecutorService() {
+ if (jobExecutorService == null){
+ jobExecutorService = new JobExecutorServiceImpl(this);
+ }
+ return jobExecutorService;
+ }
+
+ public void setJobExecutorService(JobExecutorService in_jobExecutorService) {
+ jobExecutorService = in_jobExecutorService;
+ }

 private static Log log = LogFactory.getLog(JobExecutor.class);
 }
Index: src/jpdl/org/jbpm/job/executor/JobExecutorCallback.java
===================================================================
--- src/jpdl/org/jbpm/job/executor/JobExecutorCallback.java (revision 0)
+++ src/jpdl/org/jbpm/job/executor/JobExecutorCallback.java (revision 0)
@@ -0,0 +1,18 @@
+package org.jbpm.job.executor;
+
+import org.jbpm.JbpmContext;
+
+/**
+ *
+ * @author p.pasqual
+ *
+ */
+public interface JobExecutorCallback {
+
+ /**
+ *
+ * @param in_context
+ * @return
+ */
+ public Object doInJobExecutor(JbpmContext in_jbpmContext);
+}
Index: src/jpdl/org/jbpm/job/executor/JobExecutorService.java
===================================================================
--- src/jpdl/org/jbpm/job/executor/JobExecutorService.java (revision 0)
+++ src/jpdl/org/jbpm/job/executor/JobExecutorService.java (revision 0)
@@ -0,0 +1,15 @@
+package org.jbpm.job.executor;
+
+/**
+ *
+ * @author p.pasqual
+ *
+ */
+public interface JobExecutorService {
+
+ /**
+ *
+ * @param in_callback
+ */
+ public void execute(JobExecutorCallback in_callback);
+}
Index: src/jpdl/org/jbpm/job/executor/JobExecutorServiceImpl.java
===================================================================
--- src/jpdl/org/jbpm/job/executor/JobExecutorServiceImpl.java (revision 0)
+++ src/jpdl/org/jbpm/job/executor/JobExecutorServiceImpl.java (revision 0)
@@ -0,0 +1,53 @@
+package org.jbpm.job.executor;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.jbpm.JbpmContext;
+
+import org.jbpm.persistence.JbpmPersistenceException;
+import org.jbpm.persistence.db.StaleObjectLogConfigurer;
+
+public class JobExecutorServiceImpl implements JobExecutorService {
+
+ private final JobExecutor jobExecutor;
+
+ public JobExecutorServiceImpl(JobExecutor in_jobExecutor) {
+ super();
+ jobExecutor = in_jobExecutor;
+ }
+
+ public void execute(JobExecutorCallback callback) {
+ JbpmContext jbpmContext = aquireContext();
+ try {
+ callback.doInJobExecutor(jbpmContext);
+ } finally {
+ try {
+ jbpmContext.close();
+ } catch (JbpmPersistenceException e) {
+ // if this is a stale object exception, the jbpm configuration
+ // has control over the logging
+ if ("org.hibernate.StaleObjectStateException".equals(e.getCause().getClass().getName())) {
+ log.info("problem committing execution transaction: optimistic locking failed");
+ StaleObjectLogConfigurer.staleObjectExceptionsLog.error("problem committing job execution transaction: optimistic locking failed", e);
+ } else {
+ log.error("problem committing execution transaction", e);
+ }
+ } catch (RuntimeException e) {
+ log.error("problem committing execution transaction", e);
+
+ throw e;
+ }
+ }
+ }
+
+ protected JbpmContext aquireContext(){
+ return getJobExecutor().getJbpmConfiguration().createJbpmContext();
+ }
+
+ protected JobExecutor getJobExecutor() {
+ return jobExecutor;
+ }
+
+ private static Log log = LogFactory.getLog(JobExecutorServiceImpl.class);
+
+}
Index: src/jpdl/org/jbpm/job/executor/JobExecutorThread.java
===================================================================
--- src/jpdl/org/jbpm/job/executor/JobExecutorThread.java (revision 20)
+++ src/jpdl/org/jbpm/job/executor/JobExecutorThread.java (working copy)
@@ -17,8 +17,6 @@
 import org.jbpm.db.JobSession;
 import org.jbpm.job.Job;
 import org.jbpm.job.Timer;
-import org.jbpm.persistence.JbpmPersistenceException;
-import org.jbpm.persistence.db.StaleObjectLogConfigurer;

 public class JobExecutorThread extends Thread {

@@ -153,50 +151,38 @@
 return acquiredJobs;
 }

- protected void executeJob(Job job) {
- JbpmContext jbpmContext = jbpmConfiguration.createJbpmContext();
- try {
- JobSession jobSession = jbpmContext.getJobSession();
- job = jobSession.loadJob(job.getId());
+ protected void executeJob(final Job todo) {
+ jobExecutor.getJobExecutorService().execute(new JobExecutorCallback() {
+ public Object doInJobExecutor(JbpmContext jbpmContext) {
+ JobSession jobSession = jbpmContext.getJobSession();
+ Job job = jobSession.loadJob(todo.getId());

- try {
- log.debug("executing job "+job);
- if (job.execute(jbpmContext)) {
- jobSession.deleteJob(job);
- }
+ try {
+ log.debug("executing job " + job);
+ if (job.execute(jbpmContext)) {
+ jobSession.deleteJob(job);
+ }

- } catch (Exception e) {
- log.debug("exception while executing '"+job+"'", e);
- StringWriter sw = new StringWriter();
- e.printStackTrace(new PrintWriter(sw));
- job.setException(sw.toString());
- job.setRetries(job.getRetries()-1);
- }
-
- // if this job is locked too long
- long totalLockTimeInMillis = System.currentTimeMillis() - job.getLockTime().getTime();
- if (totalLockTimeInMillis>maxLockTime) {
- jbpmContext.setRollbackOnly();
- }
+ } catch (Exception e) {
+ log.debug("exception while executing '" + job + "'", e);
+ StringWriter sw = new StringWriter();
+ e.printStackTrace(new PrintWriter(sw));
+ job.setException(sw.toString());
+ job.setRetries(job.getRetries() - 1);
+ }

- } finally {
- try {
- jbpmContext.close();
- } catch (JbpmPersistenceException e) {
- // if this is a stale object exception, the jbpm configuration has control over the logging
- if ("org.hibernate.StaleObjectStateException".equals(e.getCause().getClass().getName())) {
- log.info("problem committing job execution transaction: optimistic locking failed");
- StaleObjectLogConfigurer.staleObjectExceptionsLog.error("problem committing job execution transaction: optimistic locking failed", e);
- } else {
- log.error("problem committing job execution transaction", e);
- }
- } catch (RuntimeException e) {
- log.error("problem committing job execution transaction", e);
+ // if this job is locked too long
+ long totalLockTimeInMillis = System.currentTimeMillis() - job.getLockTime().getTime();
+ if (totalLockTimeInMillis > maxLockTime) {
+ jbpmContext.setRollbackOnly();
+ }

- throw e;
- }
- }
- }
+ return null;
+ }
+ });
+
+ }
+
 protected Date getNextDueDate() {
 Date nextDueDate = null;
 JbpmContext jbpmContext = jbpmConfiguration.createJbpmContext();

[tom.baeyens]

good proposal. last part of the draft patch is not very readable though. i'll commit addition of the setters and getters and then i'll let you know. so you can see in the cvs codebase if all access that you need is opened up.

ok ?

[fady.matar]

Does that mean decoupling the job execution from the context handling? i.e. decouple the operations on the context from the actual execution of the job?
I guess it's a good idea however I wouldn't do that using Spring since I believe that dependencies need to be minimal.
Can we add this to JIRA?

[tom.baeyens]

no. it means that they want to be able to feed in a spring based configuration file as the object factory in a JbpmConfiguration. that is already possible. but for the configuration of the job executor, some properties need to be opened up so that spring can inject those.

[fady.matar]

And what would be left to do to include a complete integration with Spring?

[p.pasqual]

hi tom,

thanks for the return, i will see if i have all acces to cvs/svn to see what you commit.

ok concerning last part of the patch it's only a refactor by externalisation of context creation/closing/error handling made by old executeJob(Job) under JobExecutorThread soo it's the same thant before. (Incremetal refactoring to limit impact of modifications).

to answer Fady: I just adapt Some Coding pattern "inspired" by Spring and i adapte to the current project without add dependencises with Spring soo dont worry. but now decoupling is sufisant to permit Sprint or Other kind of framework. Jbpm project is independant of integartion framework and it must continue to be.

About : "And what would be left to do to include a complete integration with Spring?"
it's a large domain, and there is no real answer!!! Because there is no only one complete integration with Spring or other framework but multiple coresponding to your need, your existing software, and level of integration or use of framworks.

Soo the good question can be more : what would be left to do to to permit to use Jbpm under all kind of architecture ? :-)

I think there is some hard coded things or some part of implementation that are no in corespondance with other in terme of design. But i prefer threat all those point later with concret and pre-package proposition. :-)

[tom.baeyens]

"p.pasqual" wrote:
Soo the good question can be more : what would be left to do to to permit to use Jbpm under all kind of architecture ? :-)

i think we already have a good starting point with jbpm 3. it's POJO based. but integrations with other projects like spring can be improved.

that will be included in jbpm 4. we're unifying all the different use cases that we have for wiring and a context mechanism a la seam. for those use cases where spring could be used i would like to actually test that integration so that jbpm becomes naturally integrated with typical spring usages.

[kukeltje]

What I (personally) want to prevent is re-inventing the at least a complex wheel with the jobscheduler instead of using/extending jms just because some people do not want to use an appserver.

I'm not against spring, but I've seen implementations of systems much more complex/wired just to prevent using MDB's.

[p.pasqual]

hi, tom

ok i wait your commit and after we will see for next...

regarding cvs co/commit: i'm currently behind a firewall/proxy at work soo it's very difficute to directly work with my IDE. i have found a work arround to do a remote checkout and see changes. We will see later if necessary to change things in case of multiple project colaboration/participation.


best regards
p.pasqual

[jeffj55374]

Hi,

"kukeltje" wrote:
What I (personally) want to prevent is re-inventing the at least a complex wheel with the jobscheduler instead of using/extending jms just because some people do not want to use an appserver.

I'm not against spring, but I've seen implementations of systems much more complex/wired just to prevent using MDB's.

Here's just another vote for improved spring integration. Generally I agree with your comment, but the current application I'm working on is a dedicated background application that doesn't need the overhead of an appserver so we are using jBPM in a standalone environment. This is one of the reasons we picked jBPM because of the stated goal of being lightweight and able to run in a standalone / embedded environment w/o requiring an appserver.

[kukeltje]

that doesn't need the overhead of an appserver

but you do need scalable async processing ;-)

[fady.matar]

I have included the setters for the JobExecutor. That should be sufficient to expose the properties to external frameworks. Let me know if anything else is required.