[jbpm-commits] JBoss JBPM SVN: r4073 - in jbpm3/trunk/modules: core/src/main/java/org/jbpm/job/executor and 8 other directories.

Author: alex.guizar(a)jboss.com Date: 2009-02-27 20:05:43 -0500 (Fri, 27 Feb 2009) New Revision: 4073 Added: jbpm3/trunk/modules/enterprise/src/test/java/org/jbpm/enterprise/jbpm1952/ jbpm3/trunk/modules/enterprise/src/test/java/org/jbpm/enterprise/jbpm1952/JBPM1952Test.java Removed: jbpm3/trunk/modules/enterprise/src/main/java/org/jbpm/ejb/impl/ExecuteJobsCommand.java Modified: jbpm3/trunk/modules/core/src/main/java/org/jbpm/db/JobSession.java jbpm3/trunk/modules/core/src/main/java/org/jbpm/job/executor/JobExecutorThread.java jbpm3/trunk/modules/core/src/main/java/org/jbpm/job/executor/LockMonitorThread.java jbpm3/trunk/modules/core/src/main/java/org/jbpm/persistence/db/DbPersistenceService.java jbpm3/trunk/modules/core/src/main/java/org/jbpm/persistence/jta/JtaDbPersistenceService.java jbpm3/trunk/modules/core/src/main/java/org/jbpm/svc/Services.java jbpm3/trunk/modules/core/src/test/java/org/jbpm/db/JobSessionDbTest.java jbpm3/trunk/modules/enterprise/src/main/java/org/jbpm/ejb/impl/ExecuteJobCommand.java jbpm3/trunk/modules/enterprise/src/test/java/org/jbpm/enterprise/AbstractEnterpriseTestCase.java jbpm3/trunk/modules/enterprise/src/test/java/org/jbpm/enterprise/jms/JmsMessageTest.java Log: JBPM-1952: SOSE in async fork with multiple job processors Modified: jbpm3/trunk/modules/core/src/main/java/org/jbpm/db/JobSession.java =================================================================== --- jbpm3/trunk/modules/core/src/main/java/org/jbpm/db/JobSession.java 2009-02-27 22:08:53 UTC (rev 4072) +++ jbpm3/trunk/modules/core/src/main/java/org/jbpm/db/JobSession.java 2009-02-28 01:05:43 UTC (rev 4073) @@ -28,7 +28,6 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.hibernate.Criteria; import org.hibernate.HibernateException; import org.hibernate.Query; import org.hibernate.Session; @@ -118,11 +117,13 @@ */ public List<Job> findFailedJobs() { try { - List jobs = session.createCriteria(Job.class) + List<?> jobs = session.createCriteria(Job.class) .add(Restrictions.eq("retries", 0)) - .add(Restrictions.isNotNull("exception")).list(); + .add(Restrictions.isNotNull("exception")) + .list(); return CollectionUtil.checkList(jobs, Job.class); - } catch (Exception e) { + } + catch (HibernateException e) { throw new JbpmException("couldn't find failed jobs", e); } } Modified: jbpm3/trunk/modules/core/src/main/java/org/jbpm/job/executor/JobExecutorThread.java =================================================================== --- jbpm3/trunk/modules/core/src/main/java/org/jbpm/job/executor/JobExecutorThread.java 2009-02-27 22:08:53 UTC (rev 4072) +++ jbpm3/trunk/modules/core/src/main/java/org/jbpm/job/executor/JobExecutorThread.java 2009-02-28 01:05:43 UTC (rev 4073) @@ -8,7 +8,6 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.hibernate.HibernateException; import org.jbpm.JbpmConfiguration; import org.jbpm.JbpmContext; @@ -16,8 +15,8 @@ import org.jbpm.graph.exe.ProcessInstance; import org.jbpm.job.Job; import org.jbpm.persistence.JbpmPersistenceException; +import org.jbpm.persistence.db.DbPersistenceService; import org.jbpm.persistence.db.StaleObjectLogConfigurer; -import org.jbpm.svc.Services; public class JobExecutorThread extends Thread { @@ -143,8 +142,8 @@ log.debug("obtained lock on jobs: " + acquiredJobs); } catch (JbpmPersistenceException e) { - // if this is a stale object exception, keep it quiet - if (Services.isCausedByStaleState(e)) { + // if this is a stale state exception, keep it quiet + if (DbPersistenceService.isStaleStateException(e)) { log.debug("optimistic locking failed, could not acquire jobs " + acquiredJobs); StaleObjectLogConfigurer.getStaleObjectExceptionsLog().error( "optimistic locking failed, could not acquire jobs " + acquiredJobs, e); @@ -177,7 +176,7 @@ } catch (Exception e) { log.debug("exception while executing " + job, e); - if (!isPersistenceException(e)) { + if (!DbPersistenceService.isPersistenceException(e)) { StringWriter memoryWriter = new StringWriter(); e.printStackTrace(new PrintWriter(memoryWriter)); job.setException(memoryWriter.toString()); @@ -201,7 +200,7 @@ } catch (JbpmPersistenceException e) { // if this is a stale state exception, keep it quiet - if (Services.isCausedByStaleState(e)) { + if (DbPersistenceService.isStaleStateException(e)) { log.debug("optimistic locking failed, could not complete job " + job); StaleObjectLogConfigurer.getStaleObjectExceptionsLog().error( "optimistic locking failed, could not complete job " + job, e); @@ -213,14 +212,6 @@ } } - private static boolean isPersistenceException(Throwable throwable) { - do { - if (throwable instanceof HibernateException) return true; - throwable = throwable.getCause(); - } while (throwable != null); - return false; - } - protected Date getNextDueDate() { Date nextDueDate = null; JbpmContext jbpmContext = jbpmConfiguration.createJbpmContext(); @@ -238,8 +229,8 @@ jbpmContext.close(); } catch (JbpmPersistenceException e) { - // if this is a stale object exception, keep it quiet - if (Services.isCausedByStaleState(e)) { + // if this is a stale state exception, keep it quiet + if (DbPersistenceService.isStaleStateException(e)) { log.debug("optimistic locking failed, could not return next due date: " + nextDueDate); StaleObjectLogConfigurer.getStaleObjectExceptionsLog().error( "optimistic locking failed, could not return next due date: " + nextDueDate, e); Modified: jbpm3/trunk/modules/core/src/main/java/org/jbpm/job/executor/LockMonitorThread.java =================================================================== --- jbpm3/trunk/modules/core/src/main/java/org/jbpm/job/executor/LockMonitorThread.java 2009-02-27 22:08:53 UTC (rev 4072) +++ jbpm3/trunk/modules/core/src/main/java/org/jbpm/job/executor/LockMonitorThread.java 2009-02-28 01:05:43 UTC (rev 4073) @@ -10,8 +10,8 @@ import org.jbpm.db.JobSession; import org.jbpm.job.Job; import org.jbpm.persistence.JbpmPersistenceException; +import org.jbpm.persistence.db.DbPersistenceService; import org.jbpm.persistence.db.StaleObjectLogConfigurer; -import org.jbpm.svc.Services; public class LockMonitorThread extends Thread { @@ -90,8 +90,8 @@ jbpmContext.close(); } catch (JbpmPersistenceException e) { - // if this is a stale object exception, keep it quiet - if (Services.isCausedByStaleState(e)) { + // if this is a stale state exception, keep it quiet + if (DbPersistenceService.isStaleStateException(e)) { log.debug("optimistic locking failed, could not unlock overdue jobs: " + overdueJobs); StaleObjectLogConfigurer.getStaleObjectExceptionsLog().error( "problem unlocking overdue jobs: optimistic locking failed", e); Modified: jbpm3/trunk/modules/core/src/main/java/org/jbpm/persistence/db/DbPersistenceService.java =================================================================== --- jbpm3/trunk/modules/core/src/main/java/org/jbpm/persistence/db/DbPersistenceService.java 2009-02-27 22:08:53 UTC (rev 4072) +++ jbpm3/trunk/modules/core/src/main/java/org/jbpm/persistence/db/DbPersistenceService.java 2009-02-28 01:05:43 UTC (rev 4073) @@ -27,6 +27,7 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.hibernate.HibernateException; import org.hibernate.Session; import org.hibernate.SessionFactory; import org.hibernate.StaleStateException; @@ -77,8 +78,8 @@ } public SessionFactory getSessionFactory() { - return session != null ? session.getSessionFactory() - : persistenceServiceFactory.getSessionFactory(); + return session != null ? session.getSessionFactory() : persistenceServiceFactory + .getSessionFactory(); } public Session getSession() { @@ -225,14 +226,15 @@ mustSessionBeFlushed = false; // commit does a flush anyway transaction.commit(); } - catch (StaleStateException e) { - log.info("problem committing transaction: optimistic locking failed"); - StaleObjectLogConfigurer.getStaleObjectExceptionsLog().error( - "optimistic locking failed while committing " + transaction, e); - return e; - } catch (Exception e) { - log.error("transaction commit failed", e); + if (isStaleStateException(e)) { + log.info("problem committing transaction: optimistic locking failed"); + StaleObjectLogConfigurer.getStaleObjectExceptionsLog().error( + "optimistic locking failed while committing " + transaction, e); + } + else { + log.error("transaction commit failed", e); + } return e; } return null; @@ -258,14 +260,15 @@ log.debug("flushing " + session); session.flush(); } - catch (StaleStateException e) { - log.info("problem flushing session: optimistic locking failed"); - StaleObjectLogConfigurer.getStaleObjectExceptionsLog().error( - "optimistic locking failed while flushing " + session, e); - return e; - } catch (Exception e) { - log.error("hibernate flush failed", e); + if (isStaleStateException(e)) { + log.info("problem flushing session: optimistic locking failed"); + StaleObjectLogConfigurer.getStaleObjectExceptionsLog().error( + "optimistic locking failed while flushing " + session, e); + } + else { + log.error("hibernate flush failed", e); + } return e; } } @@ -504,5 +507,19 @@ this.isTransactionEnabled = isTransactionEnabled; } + public static boolean isPersistenceException(Exception exception) { + for (Throwable t = exception; t != null; t = t.getCause()) { + if (t instanceof HibernateException) return true; + } + return false; + } + + public static boolean isStaleStateException(Exception exception) { + for (Throwable t = exception; t != null; t = t.getCause()) { + if (t instanceof StaleStateException) return true; + } + return false; + } + private static Log log = LogFactory.getLog(DbPersistenceService.class); } Modified: jbpm3/trunk/modules/core/src/main/java/org/jbpm/persistence/jta/JtaDbPersistenceService.java =================================================================== --- jbpm3/trunk/modules/core/src/main/java/org/jbpm/persistence/jta/JtaDbPersistenceService.java 2009-02-27 22:08:53 UTC (rev 4072) +++ jbpm3/trunk/modules/core/src/main/java/org/jbpm/persistence/jta/JtaDbPersistenceService.java 2009-02-28 01:05:43 UTC (rev 4073) @@ -32,7 +32,6 @@ import org.jbpm.JbpmException; import org.jbpm.persistence.db.DbPersistenceService; import org.jbpm.persistence.db.StaleObjectLogConfigurer; -import org.jbpm.svc.Services; public class JtaDbPersistenceService extends DbPersistenceService { @@ -93,8 +92,8 @@ return null; } catch (Exception e) { - if (Services.isCausedByStaleState(e)) { - log.info("optimistic locking failed, could not commit " + transaction); + if (isStaleStateException(e)) { + log.debug("optimistic locking failed, could not commit " + transaction); StaleObjectLogConfigurer.getStaleObjectExceptionsLog().error( "optimistic locking failed, could not commit " + transaction, e); } Modified: jbpm3/trunk/modules/core/src/main/java/org/jbpm/svc/Services.java =================================================================== --- jbpm3/trunk/modules/core/src/main/java/org/jbpm/svc/Services.java 2009-02-27 22:08:53 UTC (rev 4072) +++ jbpm3/trunk/modules/core/src/main/java/org/jbpm/svc/Services.java 2009-02-28 01:05:43 UTC (rev 4073) @@ -30,7 +30,6 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.hibernate.StaleStateException; import org.jbpm.JbpmContext; import org.jbpm.JbpmException; import org.jbpm.graph.exe.ProcessInstance; @@ -38,6 +37,7 @@ import org.jbpm.msg.MessageService; import org.jbpm.persistence.JbpmPersistenceException; import org.jbpm.persistence.PersistenceService; +import org.jbpm.persistence.db.DbPersistenceService; import org.jbpm.persistence.db.StaleObjectLogConfigurer; import org.jbpm.scheduler.SchedulerService; import org.jbpm.security.AuthenticationService; @@ -243,9 +243,8 @@ service.close(); } catch (JbpmPersistenceException e) { - // if this is a stale state exception, the jbpm configuration has control over the - // logging - if (isCausedByStaleState(e)) { + // if this is a stale state exception, keep it quiet + if (DbPersistenceService.isStaleStateException(e)) { log.info("optimistic locking failed, could not close service: " + serviceName); StaleObjectLogConfigurer.getStaleObjectExceptionsLog().error( "optimistic locking failed, could not close service: " + serviceName, e); @@ -274,13 +273,6 @@ } } - public static boolean isCausedByStaleState(Exception exception) { - for (Throwable cause = exception.getCause(); cause != null; cause = cause.getCause()) { - if (cause instanceof StaleStateException) return true; - } - return false; - } - public static void assignId(Object object) { JbpmContext jbpmContext = JbpmContext.getCurrentJbpmContext(); if (jbpmContext != null) { Modified: jbpm3/trunk/modules/core/src/test/java/org/jbpm/db/JobSessionDbTest.java =================================================================== --- jbpm3/trunk/modules/core/src/test/java/org/jbpm/db/JobSessionDbTest.java 2009-02-27 22:08:53 UTC (rev 4072) +++ jbpm3/trunk/modules/core/src/test/java/org/jbpm/db/JobSessionDbTest.java 2009-02-28 01:05:43 UTC (rev 4073) @@ -36,18 +36,18 @@ public static class FailingAction implements ActionHandler { private static final long serialVersionUID = 1L; - public void execute(ExecutionContext executionContext) throws Exception - { + + public void execute(ExecutionContext executionContext) throws Exception { throw new RuntimeException("TEST-EXCEPTION"); } } /** * Test case which generates a {@link Job} ({@link ExecuteNodeJob} via async=true) - * which causes an excpetion. Afterwards it is checked if this job + * which causes an exception. Afterwards it is checked if this job * is found by the getFailedJobs method */ - public void notestLoadFailedJobs() throws Exception { + public void testLoadFailedJobs() throws Exception { String xml = "<process-definition name='TestJob'>" + " <start-state>" @@ -64,14 +64,14 @@ // create a process definition ProcessDefinition processDefinition = ProcessDefinition.parseXmlString(xml); // save it in the database - graphSession.saveProcessDefinition(processDefinition); + graphSession.deployProcessDefinition(processDefinition); newTransaction(); - try { // start a new process instance and signal it ProcessInstance pi = jbpmContext.newProcessInstance("TestJob"); pi.getRootToken().signal(); + jbpmContext.save(pi); newTransaction(); @@ -81,15 +81,23 @@ // start job executor wait for job to be executed // and failure is written to database - startJobExecutor(); - long startTime = System.currentTimeMillis(); - while( jobSession.findFailedJobs().size() <= 0 ) { - if (System.currentTimeMillis() - startTime > timeout) { - fail("test execution exceeded treshold of " + timeout + " milliseconds"); + commitAndCloseSession(); + try { + startJobExecutor(); + + long startTime = System.currentTimeMillis(); + while( getFailedJobCount() <= 0 ) { + if (System.currentTimeMillis() - startTime > timeout) { + fail("test execution exceeded threshold of " + timeout + " milliseconds"); + } + Thread.sleep(500); } - Thread.sleep(500); } - stopJobExecutor(); + finally { + stopJobExecutor(); + beginSessionTransaction(); + } + List<Job> failedJobs = jobSession.findFailedJobs(); // now the one job we have should be failed @@ -97,16 +105,28 @@ assertEquals(1, failedJobs.size()); // and information is set on the job - assertEquals(0, failedJobs.get(0).getRetries()); - assertNotNull(failedJobs.get(0).getException() + "==null", - failedJobs.get(0).getException()); - assertTrue(failedJobs.get(0).getException() + " contains TEST-EXCEPTION", - failedJobs.get(0).getException().indexOf("TEST-EXCEPTION")>0); + Job job = failedJobs.get(0); + assertEquals(0, job.getRetries()); + + String exception = job.getException(); + assertNotNull("expected job.exception not to be null", exception); + assertTrue("expected job.exception to contain TEST-EXCEPTION", + exception.contains("TEST-EXCEPTION")); } finally { - newTransaction(); // cleanup jbpmContext.getGraphSession().deleteProcessDefinition(processDefinition.getId()); } } + + int getFailedJobCount() { + beginSessionTransaction(); + try { + return jobSession.findFailedJobs().size(); + } + finally { + commitAndCloseSession(); + } + } + } \ No newline at end of file Modified: jbpm3/trunk/modules/enterprise/src/main/java/org/jbpm/ejb/impl/ExecuteJobCommand.java =================================================================== --- jbpm3/trunk/modules/enterprise/src/main/java/org/jbpm/ejb/impl/ExecuteJobCommand.java 2009-02-27 22:08:53 UTC (rev 4072) +++ jbpm3/trunk/modules/enterprise/src/main/java/org/jbpm/ejb/impl/ExecuteJobCommand.java 2009-02-28 01:05:43 UTC (rev 4073) @@ -23,27 +23,19 @@ import java.io.PrintWriter; import java.io.StringWriter; -import java.util.List; -import javax.jms.Destination; -import javax.jms.JMSException; -import javax.jms.Message; -import javax.jms.MessageProducer; -import javax.jms.Session; -import javax.naming.NamingException; - import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.jbpm.JbpmContext; +import org.jbpm.JbpmException; import org.jbpm.command.Command; -import org.jbpm.db.JobSession; +import org.jbpm.graph.exe.ProcessInstance; import org.jbpm.job.Job; -import org.jbpm.msg.jms.JmsMessageService; -import org.jbpm.msg.jms.JmsMessageServiceFactory; -import org.jbpm.svc.Services; +import org.jbpm.persistence.db.DbPersistenceService; /** * Individual job processing command. + * * @author Alejandro Guizar */ public class ExecuteJobCommand implements Command { @@ -57,83 +49,50 @@ } public Object execute(JbpmContext jbpmContext) throws Exception { - JobSession jobSession = jbpmContext.getJobSession(); - Job job = jobSession.getJob(jobId); - if (job == null) { - log.debug("job " + jobId + " was deleted"); - return null; - } - String lockOwner = job.getLockOwner(); - if (lockOwner != null) { - log.debug(job + " is locked by " + lockOwner); - return null; - } - lockOwner = Long.toString(jobId); + Job job = acquireJob(jbpmContext); + executeJob(job, jbpmContext); + return job; + } + + private Job acquireJob(JbpmContext jbpmContext) { + Job job = jbpmContext.getJobSession().loadJob(jobId); + + // register process instance for automatic save + // see https://jira.jboss.org/jira/browse/JBPM-1015 + ProcessInstance processInstance = job.getProcessInstance(); + jbpmContext.addAutoSaveProcessInstance(processInstance); + + // if job is exclusive, lock process instance if (job.isExclusive()) { - List<Job> exclusiveJobs = jobSession.findExclusiveJobs(lockOwner, job.getProcessInstance()); - // lock exclusive jobs - if (exclusiveJobs.isEmpty()) { - // may happen if isolation level is below repeatable read - log.debug(job + " was locked during attempt to lock other jobs"); - return null; - } - long[] exclusiveJobIds = new long[exclusiveJobs.size()]; - int i = 0; - for (Job exclusiveJob : exclusiveJobs) { - exclusiveJob.setLockOwner(lockOwner); - exclusiveJobIds[i++] = exclusiveJob.getId(); - } - log.debug("locking jobs " + exclusiveJobIds); - // execute exclusive jobs in separate transaction - postJobsExecution(jbpmContext, exclusiveJobIds); + jbpmContext.getGraphSession().lockProcessInstance(processInstance); } - else { - // lock job to prevent others from deleting it - job.setLockOwner(lockOwner); - log.debug("executing " + job); - executeJob(job, jbpmContext); - } - return null; + + // mark job as locked to prevent other parts of the engine from deleting it + job.setLockOwner(toString()); + return job; } static void executeJob(Job job, JbpmContext jbpmContext) { + log.debug("executing " + job); try { if (job.execute(jbpmContext)) { jbpmContext.getJobSession().deleteJob(job); } } - catch (RuntimeException e) { - // nothing to do but clean up and exit - throw e; - } catch (Exception e) { - // save data about recoverable error condition - log.error("exception while executing " + job, e); - StringWriter memoryWriter = new StringWriter(); - e.printStackTrace(new PrintWriter(memoryWriter)); - job.setException(memoryWriter.toString()); - job.setRetries(job.getRetries() - 1); + log.debug("exception while executing " + job, e); + if (!DbPersistenceService.isPersistenceException(e)) { + StringWriter memoryWriter = new StringWriter(); + e.printStackTrace(new PrintWriter(memoryWriter)); + job.setException(memoryWriter.toString()); + } + else { + // allowing a transaction to proceed after a persistence exception is unsafe + throw e instanceof RuntimeException ? (RuntimeException) e : + new JbpmException("failed to execute " + job, e); + } } } - private static void postJobsExecution(JbpmContext jbpmContext, long[] exclusiveJobIds) - throws NamingException, JMSException { - Services services = jbpmContext.getServices(); - JmsMessageServiceFactory messageServiceFactory = (JmsMessageServiceFactory) services.getServiceFactory(Services.SERVICENAME_MESSAGE); - Destination destination = messageServiceFactory.getCommandDestination(); - - JmsMessageService messageService = (JmsMessageService) services.getMessageService(); - Session session = messageService.getSession(); - MessageProducer producer = session.createProducer(destination); - try { - Command command = new ExecuteJobsCommand(exclusiveJobIds); - Message message = session.createObjectMessage(command); - producer.send(message); - } - finally { - producer.close(); - } - } - private static Log log = LogFactory.getLog(ExecuteJobCommand.class); } Deleted: jbpm3/trunk/modules/enterprise/src/main/java/org/jbpm/ejb/impl/ExecuteJobsCommand.java =================================================================== --- jbpm3/trunk/modules/enterprise/src/main/java/org/jbpm/ejb/impl/ExecuteJobsCommand.java 2009-02-27 22:08:53 UTC (rev 4072) +++ jbpm3/trunk/modules/enterprise/src/main/java/org/jbpm/ejb/impl/ExecuteJobsCommand.java 2009-02-28 01:05:43 UTC (rev 4073) @@ -1,57 +0,0 @@ -/* - * JBoss, Home of Professional Open Source - * Copyright 2005, JBoss Inc., and individual contributors as indicated - * by the @authors tag. See the copyright.txt in the distribution for a - * full listing of individual contributors. - * - * This is free software; you can redistribute it and/or modify it - * under the terms of the GNU Lesser General Public License as - * published by the Free Software Foundation; either version 2.1 of - * the License, or (at your option) any later version. - * - * This software is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU - * Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public - * License along with this software; if not, write to the Free - * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA - * 02110-1301 USA, or see the FSF site: http://www.fsf.org. - */ -package org.jbpm.ejb.impl; - -import java.util.Arrays; -import java.util.List; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import org.jbpm.JbpmContext; -import org.jbpm.command.Command; -import org.jbpm.job.Job; - -/** - * Batch job processing command. - * @author Alejandro Guizar - */ -public class ExecuteJobsCommand implements Command { - - private long[] jobIds; - - private static final long serialVersionUID = 1L; - private static Log log = LogFactory.getLog(ExecuteJobsCommand.class); - - public ExecuteJobsCommand(long[] jobIds) { - this.jobIds = jobIds; - } - - public Object execute(JbpmContext jbpmContext) throws Exception { - log.debug("executing jobs " + Arrays.toString(jobIds)); - List<Job> jobs = jbpmContext.getJobSession().loadJobs(jobIds); - for (Job job : jobs) { - job.execute(jbpmContext); - } - return null; - } - -} Modified: jbpm3/trunk/modules/enterprise/src/test/java/org/jbpm/enterprise/AbstractEnterpriseTestCase.java =================================================================== --- jbpm3/trunk/modules/enterprise/src/test/java/org/jbpm/enterprise/AbstractEnterpriseTestCase.java 2009-02-27 22:08:53 UTC (rev 4072) +++ jbpm3/trunk/modules/enterprise/src/test/java/org/jbpm/enterprise/AbstractEnterpriseTestCase.java 2009-02-28 01:05:43 UTC (rev 4073) @@ -46,6 +46,7 @@ import org.jbpm.command.SignalCommand; import org.jbpm.command.StartProcessInstanceCommand; import org.jbpm.ejb.LocalCommandServiceHome; +import org.jbpm.graph.def.Event; import org.jbpm.graph.def.EventCallback; import org.jbpm.graph.def.ProcessDefinition; import org.jbpm.graph.exe.ProcessInstance; @@ -63,6 +64,7 @@ private static Destination commandQueue; private static ConnectionFactory jmsConnectionFactory; + private static final int MAX_WAIT_TIME = 30 * 1000; private static final Log log = LogFactory.getLog(AbstractEnterpriseTestCase.class); protected AbstractEnterpriseTestCase() { @@ -121,10 +123,19 @@ return processInstance.hasEnded(); } + protected boolean waitForProcessInstanceEnd(long processInstanceId) { + long startTime = System.currentTimeMillis(); + do { + EventCallback.waitForEvent(Event.EVENTTYPE_PROCESS_END, 1000); + if (System.currentTimeMillis() - startTime > MAX_WAIT_TIME) return false; + } while (!hasProcessInstanceEnded(processInstanceId)); + return true; + } + protected Object getVariable(final long processInstanceId, final String variableName) { return commandService.execute(new Command() { private static final long serialVersionUID = 1L; - + public Object execute(JbpmContext jbpmContext) throws Exception { ProcessInstance processInstance = jbpmContext.loadProcessInstance(processInstanceId); return processInstance.getContextInstance().getVariable(variableName); Added: jbpm3/trunk/modules/enterprise/src/test/java/org/jbpm/enterprise/jbpm1952/JBPM1952Test.java =================================================================== --- jbpm3/trunk/modules/enterprise/src/test/java/org/jbpm/enterprise/jbpm1952/JBPM1952Test.java (rev 0) +++ jbpm3/trunk/modules/enterprise/src/test/java/org/jbpm/enterprise/jbpm1952/JBPM1952Test.java 2009-02-28 01:05:43 UTC (rev 4073) @@ -0,0 +1,72 @@ +package org.jbpm.enterprise.jbpm1952; + +import junit.framework.Test; + +import org.jboss.bpm.api.test.IntegrationTestSetup; +import org.jbpm.enterprise.AbstractEnterpriseTestCase; + +/** + * Use JMS instead of DBMS for storing jobs, so that each job is not taken by multiple job executor + * threads. + * + * @see <a href="https://jira.jboss.org/jira/browse/JBPM-1952">JBPM-195... + * @author Alejandro Guizar + */ +public class JBPM1952Test extends AbstractEnterpriseTestCase { + + private static final int PROCESS_EXECUTION_COUNT = 20; + + public static Test suite() { + return new IntegrationTestSetup(JBPM1952Test.class, "enterprise-test.war"); + } + + public void testStaleStateInAsyncFork() { + deployProcessDefinition("<process-definition name='jbpm1952'>" + + " <event type='process-end'>" + + " <action expression='#{eventCallback.processEnd}' />" + + " </event>" + + " <start-state>" + + " <transition to='a' />" + + " </start-state>" + + " <node name='a' async='true'>" + + " <transition to='b' />" + + " </node>" + + " <node name='b' async='true'>" + + " <transition to='fork' />" + + " </node>" + + " <fork name='fork'>" + + " <transition to='c1' name='to_c1'/>" + + " <transition to='c2' name='to_c2'/>" + + " <transition to='c3' name='to_c3'/>" + + " </fork>" + + " <node name='c1' async='true'>" + + " <transition to='join' />" + + " </node>" + + " <node name='c2' async='true'>" + + " <transition to='join' />" + + " </node>" + + " <node name='c3' async='true'>" + + " <transition to='join' />" + + " </node>" + + " <join name='join' async='exclusive'>" + + " <transition to='d' />" + + " </join>" + + " <node name='d' async='true'>" + + " <transition to='end' />" + + " </node>" + + " <end-state name='end'/>" + + "</process-definition>"); + + long[] processInstanceIds = new long[PROCESS_EXECUTION_COUNT]; + for (int i = 0; i < PROCESS_EXECUTION_COUNT; i++) { + processInstanceIds[i] = startProcessInstance("jbpm1952").getId(); + } + + for (int i = 0; i < PROCESS_EXECUTION_COUNT; i++) { + long processInstanceId = processInstanceIds[i]; + assertTrue("expected process instance " + processInstanceId + " to have ended", + waitForProcessInstanceEnd(processInstanceId)); + } + } + +} Modified: jbpm3/trunk/modules/enterprise/src/test/java/org/jbpm/enterprise/jms/JmsMessageTest.java =================================================================== --- jbpm3/trunk/modules/enterprise/src/test/java/org/jbpm/enterprise/jms/JmsMessageTest.java 2009-02-27 22:08:53 UTC (rev 4072) +++ jbpm3/trunk/modules/enterprise/src/test/java/org/jbpm/enterprise/jms/JmsMessageTest.java 2009-02-28 01:05:43 UTC (rev 4073) @@ -36,8 +36,7 @@ */ public class JmsMessageTest extends AbstractEnterpriseTestCase { - private static final int maxWaitTime = 10 * 1000; - private static final int processExecutionCount = 5; + private static final int PROCESS_INSTANCE_COUNT = 5; public static Test suite() throws Exception { return new IntegrationTestSetup(JmsMessageTest.class, "enterprise-test.war"); @@ -201,29 +200,18 @@ + " <end-state name='end' />" + "</process-definition>"); - long[] processInstanceIds = new long[processExecutionCount]; - for (int i = 0; i < processExecutionCount; i++) { + long[] processInstanceIds = new long[PROCESS_INSTANCE_COUNT]; + for (int i = 0; i < PROCESS_INSTANCE_COUNT; i++) { processInstanceIds[i] = startProcessInstance("execution").getId(); EventCallback.waitForEvent(Event.EVENTTYPE_NODE_ENTER); } - for (int i = 0; i < processExecutionCount; i++) { + for (int i = 0; i < PROCESS_INSTANCE_COUNT; i++) { EventCallback.waitForEvent(Event.EVENTTYPE_NODE_LEAVE); } - for (int i = 0; i < processExecutionCount; i++) { + for (int i = 0; i < PROCESS_INSTANCE_COUNT; i++) { long processInstanceId = processInstanceIds[i]; assertTrue("expected process instance " + processInstanceId + " to have ended", waitForProcessInstanceEnd(processInstanceId)); } } - - private boolean waitForProcessInstanceEnd(long processInstanceId) { - long startTime = System.currentTimeMillis(); - do { - EventCallback.waitForEvent(Event.EVENTTYPE_PROCESS_END, 1000); - if (System.currentTimeMillis() - startTime > maxWaitTime) { - return false; - } - } while (!hasProcessInstanceEnded(processInstanceId)); - return true; - } }