[jbpm-commits] JBoss JBPM SVN: r5999 - in jbpm3/branches/jbpm-3.2-soa/modules/core/src: main/java/org/jbpm/job/executor and 6 other directories.

Author: alex.guizar(a)jboss.com Date: 2009-12-18 16:13:44 -0500 (Fri, 18 Dec 2009) New Revision: 5999 Added: jbpm3/branches/jbpm-3.2-soa/modules/core/src/test/java/org/jbpm/jbpm2691/ jbpm3/branches/jbpm-3.2-soa/modules/core/src/test/java/org/jbpm/jbpm2691/BombAction.java jbpm3/branches/jbpm-3.2-soa/modules/core/src/test/java/org/jbpm/jbpm2691/JBPM2691Test.java jbpm3/branches/jbpm-3.2-soa/modules/core/src/test/resources/org/jbpm/jbpm2691/ jbpm3/branches/jbpm-3.2-soa/modules/core/src/test/resources/org/jbpm/jbpm2691/gpd.xml jbpm3/branches/jbpm-3.2-soa/modules/core/src/test/resources/org/jbpm/jbpm2691/processdefinition.xml Modified: jbpm3/branches/jbpm-3.2-soa/modules/core/src/main/java/org/jbpm/db/AbstractDbTestCase.java jbpm3/branches/jbpm-3.2-soa/modules/core/src/main/java/org/jbpm/job/executor/JobExecutorThread.java jbpm3/branches/jbpm-3.2-soa/modules/core/src/main/resources/org/jbpm/db/hibernate.queries.hbm.xml jbpm3/branches/jbpm-3.2-soa/modules/core/src/test/resources/org/jbpm/jbpm2574/processdefinition.xml Log: JBPM-2691: Save exceptions thrown while executing a job in a separate transaction Modified: jbpm3/branches/jbpm-3.2-soa/modules/core/src/main/java/org/jbpm/db/AbstractDbTestCase.java =================================================================== --- jbpm3/branches/jbpm-3.2-soa/modules/core/src/main/java/org/jbpm/db/AbstractDbTestCase.java 2009-12-18 19:00:53 UTC (rev 5998) +++ jbpm3/branches/jbpm-3.2-soa/modules/core/src/main/java/org/jbpm/db/AbstractDbTestCase.java 2009-12-18 21:13:44 UTC (rev 5999) @@ -79,7 +79,8 @@ } private void ensureCleanDatabase() { - DbPersistenceServiceFactory persistenceServiceFactory = (DbPersistenceServiceFactory) getJbpmConfiguration().getServiceFactory("persistence"); + DbPersistenceServiceFactory persistenceServiceFactory = + (DbPersistenceServiceFactory) getJbpmConfiguration().getServiceFactory("persistence"); if (persistenceServiceFactory == null) return; boolean hasLeftOvers = false; @@ -91,8 +92,8 @@ Long count = (Long) entry.getValue(); if (count.intValue() != 0) { hasLeftOvers = true; - System.err.println("FIXME: " + getClass().getName() + "." + getName() + " left " - + count + " records in " + entry.getKey()); + System.err.println("FIXME: " + getClass().getName() + "." + getName() + + " left " + count + " records in " + entry.getKey()); } } @@ -102,8 +103,10 @@ } protected String getHibernateDialect() { - DbPersistenceServiceFactory persistenceServiceFactory = (DbPersistenceServiceFactory) jbpmContext.getServiceFactory(Services.SERVICENAME_PERSISTENCE); - return persistenceServiceFactory.getConfiguration().getProperty(Environment.DIALECT); + DbPersistenceServiceFactory persistenceServiceFactory = + (DbPersistenceServiceFactory) jbpmContext.getServiceFactory(Services.SERVICENAME_PERSISTENCE); + return persistenceServiceFactory.getConfiguration() + .getProperty(Environment.DIALECT); } protected void beginSessionTransaction() { @@ -130,7 +133,8 @@ protected TaskInstance saveAndReload(TaskInstance taskInstance) { jbpmContext.save(taskInstance); newTransaction(); - return (TaskInstance) session.load(TaskInstance.class, new Long(taskInstance.getId())); + return (TaskInstance) session.load(TaskInstance.class, new Long( + taskInstance.getId())); } protected ProcessDefinition saveAndReload(ProcessDefinition pd) { @@ -186,7 +190,7 @@ } protected void waitForJobs(final long timeout) { - final long startTime = System.currentTimeMillis(); + long startTime = System.currentTimeMillis(); int previousCount = 0; long previousTime = 0L; @@ -194,9 +198,9 @@ long waitPeriod = 500; for (int currentCount; (currentCount = getNbrOfJobsAvailable()) > 0;) { - final long currentTime = System.currentTimeMillis(); + long currentTime = System.currentTimeMillis(); - final long elapsedTime = currentTime - startTime; + long elapsedTime = currentTime - startTime; if (elapsedTime > timeout) { fail("test execution exceeded threshold of " + timeout + " ms"); } @@ -214,11 +218,12 @@ waitPeriod = 5000; } else { - final long remainingTime = timeout - elapsedTime; + long remainingTime = timeout - elapsedTime; if (waitPeriod > remainingTime) waitPeriod = remainingTime; } - log.debug("waiting " + waitPeriod + " ms for " + currentCount + " jobs to be executed"); + log.debug("waiting " + waitPeriod + " ms for " + currentCount + + " jobs to be executed"); try { Thread.sleep(waitPeriod); } @@ -233,12 +238,12 @@ protected int getNbrOfJobsAvailable() { if (session != null) { - return getNbrOfJobsAvailable(session); + return getJobCount(session); } else { beginSessionTransaction(); try { - return getNbrOfJobsAvailable(session); + return getJobCount(session); } finally { commitAndCloseSession(); @@ -246,16 +251,15 @@ } } - private int getNbrOfJobsAvailable(Session session) { - Number jobs = (Number) session.createQuery("select count(*) from org.jbpm.job.Job") - .uniqueResult(); - return jobs.intValue(); + private int getJobCount(Session session) { + Number jobCount = (Number) session.createQuery("select count(*) " + + "from org.jbpm.job.Job where retries > 0").uniqueResult(); + return jobCount.intValue(); } protected int getTimerCount() { - Number timerCount = (Number) session.createQuery("select count(*) from org.jbpm.job.Timer") - .uniqueResult(); - log.debug("there are " + timerCount + " timers in the database"); + Number timerCount = (Number) session.createQuery("select count(*) " + + "from org.jbpm.job.Timer where retries > 0").uniqueResult(); return timerCount.intValue(); } Modified: jbpm3/branches/jbpm-3.2-soa/modules/core/src/main/java/org/jbpm/job/executor/JobExecutorThread.java =================================================================== --- jbpm3/branches/jbpm-3.2-soa/modules/core/src/main/java/org/jbpm/job/executor/JobExecutorThread.java 2009-12-18 19:00:53 UTC (rev 5998) +++ jbpm3/branches/jbpm-3.2-soa/modules/core/src/main/java/org/jbpm/job/executor/JobExecutorThread.java 2009-12-18 21:13:44 UTC (rev 5999) @@ -43,8 +43,8 @@ * @deprecated use {@link #JobExecutorThread(String, JobExecutor)} instead */ public JobExecutorThread(String name, JobExecutor jobExecutor, - JbpmConfiguration jbpmConfiguration, int idleInterval, int maxIdleInterval, - long maxLockTime, int maxHistory) { + JbpmConfiguration jbpmConfiguration, int idleInterval, + int maxIdleInterval, long maxLockTime, int maxHistory) { super(name); this.jobExecutor = jobExecutor; this.jbpmConfiguration = jbpmConfiguration; @@ -60,7 +60,12 @@ Collection acquiredJobs = acquireJobs(); for (Iterator i = acquiredJobs.iterator(); i.hasNext() && isActive;) { Job job = (Job) i.next(); - executeJob(job); + try { + executeJob(job); + } + catch (Exception e) { + saveJobException(job, e); + } } if (isActive) { long waitPeriod = getWaitPeriod(currentIdleInterval); @@ -75,21 +80,23 @@ } catch (RuntimeException e) { if (isActive) { - log.error("exception in " + getName() + ", waiting " + currentIdleInterval + " ms", e); + log.error("exception in " + getName() + ", waiting " + + currentIdleInterval + " ms", e); try { synchronized (jobExecutor) { jobExecutor.wait(currentIdleInterval); } + // after an exception, double the current idle interval to avoid + // throwing exceptions continuously during anomalous conditions + currentIdleInterval *= 2; + if (currentIdleInterval > maxIdleInterval + || currentIdleInterval < 0) { + currentIdleInterval = maxIdleInterval; + } } catch (InterruptedException ie) { log.debug(getName() + " got interrupted: " + e.getMessage()); } - // after an exception, double the current idle interval to prevent - // continuous exception throwing while an anomalous condition prevails - currentIdleInterval *= 2; - if (currentIdleInterval > maxIdleInterval || currentIdleInterval < 0) { - currentIdleInterval = maxIdleInterval; - } } } catch (InterruptedException e) { @@ -115,7 +122,8 @@ if (job.isExclusive()) { ProcessInstance processInstance = job.getProcessInstance(); log.debug("finding exclusive jobs for " + processInstance); - jobsToLock = jobSession.findExclusiveJobs(lockOwner, processInstance); + jobsToLock = + jobSession.findExclusiveJobs(lockOwner, processInstance); log.debug("acquiring " + jobsToLock + " for " + processInstance); } else { @@ -157,7 +165,7 @@ return acquiredJobs; } - protected void executeJob(Job job) { + protected void executeJob(Job job) throws Exception { JbpmContext jbpmContext = jbpmConfiguration.createJbpmContext(); try { JobSession jobSession = jbpmContext.getJobSession(); @@ -174,17 +182,13 @@ } } catch (Exception e) { - log.debug("exception while executing " + job, e); - if (!DbPersistenceService.isPersistenceException(e)) { - StringWriter memoryWriter = new StringWriter(); - e.printStackTrace(new PrintWriter(memoryWriter)); - job.setException(memoryWriter.toString()); - job.setRetries(job.getRetries() - 1); - } - else { - // prevent unsafe use of the session after an exception occurs - jbpmContext.setRollbackOnly(); - } + // prevent unsafe use of the session after an exception occurs + jbpmContext.setRollbackOnly(); + + if (!DbPersistenceService.isLockingException(e)) throw e; + // if this is a locking exception, keep it quiet + StaleObjectLogConfigurer.getStaleObjectExceptionsLog() + .error("failed to complete " + job); } catch (Error e) { jbpmContext.setRollbackOnly(); @@ -192,7 +196,8 @@ } // if this job is locked too long - long totalLockTimeInMillis = System.currentTimeMillis() - job.getLockTime().getTime(); + long totalLockTimeInMillis = + System.currentTimeMillis() - job.getLockTime().getTime(); if (totalLockTimeInMillis > maxLockTime) { jbpmContext.setRollbackOnly(); } @@ -204,12 +209,39 @@ catch (RuntimeException e) { if (!DbPersistenceService.isLockingException(e)) throw e; // if this is a locking exception, keep it quiet - StaleObjectLogConfigurer.getStaleObjectExceptionsLog().error("failed to complete job " - + job); + StaleObjectLogConfigurer.getStaleObjectExceptionsLog() + .error("failed to complete " + job); } } } + private void saveJobException(Job job, Exception exception) { + JbpmContext jbpmContext = jbpmConfiguration.createJbpmContext(); + try { + JobSession jobSession = jbpmContext.getJobSession(); + // not reattaching existing job as it may contain updates + job = jobSession.loadJob(job.getId()); + + StringWriter out = new StringWriter(); + exception.printStackTrace(new PrintWriter(out)); + job.setException(out.toString()); + job.setRetries(job.getRetries() - 1); + } + catch (RuntimeException e) { + jbpmContext.setRollbackOnly(); + log.error("failed to save job exception", exception); + throw e; + } + catch (Error e) { + jbpmContext.setRollbackOnly(); + log.error("failed to save job exception", exception); + throw e; + } + finally { + jbpmContext.close(); + } + } + protected long getWaitPeriod(int currentIdleInterval) { long waitPeriod = currentIdleInterval; @@ -255,7 +287,7 @@ if (!DbPersistenceService.isLockingException(e)) throw e; // if this is a locking exception, keep it quiet StaleObjectLogConfigurer.getStaleObjectExceptionsLog() - .error("failed to determine next due date for job executor thread " + lockOwner); + .error("failed to determine next due date"); nextDueDate = null; } } @@ -270,7 +302,8 @@ } /** - * Indicates that this thread should stop running. Execution will cease shortly afterwards. + * Indicates that this thread should stop running. Execution will cease + * shortly afterwards. */ public void deactivate() { if (isActive) { Modified: jbpm3/branches/jbpm-3.2-soa/modules/core/src/main/resources/org/jbpm/db/hibernate.queries.hbm.xml =================================================================== --- jbpm3/branches/jbpm-3.2-soa/modules/core/src/main/resources/org/jbpm/db/hibernate.queries.hbm.xml 2009-12-18 19:00:53 UTC (rev 5998) +++ jbpm3/branches/jbpm-3.2-soa/modules/core/src/main/resources/org/jbpm/db/hibernate.queries.hbm.xml 2009-12-18 21:13:44 UTC (rev 5999) @@ -265,7 +265,7 @@ <![CDATA[ select job from org.jbpm.job.Job as job - where ( (job.lockOwner is null) or (job.lockOwner = :lockOwner) ) + where (job.lockOwner is null or job.lockOwner = :lockOwner) and job.retries > 0 and job.dueDate <= :now and job.isSuspended = false @@ -277,7 +277,7 @@ <![CDATA[ select job from org.jbpm.job.Job as job - where ( (job.lockOwner is null) or (job.lockOwner = :lockOwner) ) + where (job.lockOwner is null or job.lockOwner = :lockOwner) and job.retries > 0 and job.dueDate <= :now and job.processInstance = :processInstance @@ -291,7 +291,7 @@ <![CDATA[ select job from org.jbpm.job.Job as job - where ( (job.lockOwner is null) or (job.lockOwner = :lockOwner) ) + where (job.lockOwner is null or job.lockOwner = :lockOwner) and job.retries > 0 and job.isSuspended = false order by job.dueDate asc @@ -302,8 +302,8 @@ <![CDATA[ select job from org.jbpm.job.Job as job - where ( (job.lockOwner is null) or (job.lockOwner = :lockOwner) ) - and job.id not in ( :monitoredJobIds ) + where (job.lockOwner is null or job.lockOwner = :lockOwner) + and job.id not in (:monitoredJobIds) and job.retries > 0 and job.isSuspended = false order by job.dueDate asc @@ -406,7 +406,7 @@ <![CDATA[ select ti from org.jbpm.taskmgmt.exe.TaskInstance as ti - where ti.actorId in ( :actorIds ) + where ti.actorId in (:actorIds) and ti.isSuspended = false and ti.isOpen = true ]]> @@ -429,7 +429,7 @@ select distinct ti.id from org.jbpm.taskmgmt.exe.TaskInstance ti join ti.pooledActors pooledActor - where pooledActor.actorId in ( :actorIds ) + where pooledActor.actorId in (:actorIds) and ti.actorId is null and ti.isSuspended = false and ti.isOpen = true @@ -460,7 +460,7 @@ <![CDATA[ select ti from org.jbpm.taskmgmt.exe.TaskInstance ti - where ti.id in ( :taskInstanceIds ) + where ti.id in (:taskInstanceIds) ]]> </query> Added: jbpm3/branches/jbpm-3.2-soa/modules/core/src/test/java/org/jbpm/jbpm2691/BombAction.java =================================================================== --- jbpm3/branches/jbpm-3.2-soa/modules/core/src/test/java/org/jbpm/jbpm2691/BombAction.java (rev 0) +++ jbpm3/branches/jbpm-3.2-soa/modules/core/src/test/java/org/jbpm/jbpm2691/BombAction.java 2009-12-18 21:13:44 UTC (rev 5999) @@ -0,0 +1,14 @@ +package org.jbpm.jbpm2691; + +import org.jbpm.graph.def.ActionHandler; +import org.jbpm.graph.exe.ExecutionContext; + +public class BombAction implements ActionHandler { + + private static final long serialVersionUID = 1L; + + public void execute(ExecutionContext executionContext) throws Exception { + throw new Exception("boom"); + } + +} Property changes on: jbpm3/branches/jbpm-3.2-soa/modules/core/src/test/java/org/jbpm/jbpm2691/BombAction.java ___________________________________________________________________ Name: svn:keywords + Id Revision Name: svn:eol-style + native Added: jbpm3/branches/jbpm-3.2-soa/modules/core/src/test/java/org/jbpm/jbpm2691/JBPM2691Test.java =================================================================== --- jbpm3/branches/jbpm-3.2-soa/modules/core/src/test/java/org/jbpm/jbpm2691/JBPM2691Test.java (rev 0) +++ jbpm3/branches/jbpm-3.2-soa/modules/core/src/test/java/org/jbpm/jbpm2691/JBPM2691Test.java 2009-12-18 21:13:44 UTC (rev 5999) @@ -0,0 +1,74 @@ +/* + * 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.jbpm2691; + +import java.util.List; + +import org.jbpm.db.AbstractDbTestCase; +import org.jbpm.graph.def.ProcessDefinition; +import org.jbpm.graph.exe.ProcessInstance; +import org.jbpm.graph.exe.Token; +import org.jbpm.job.Job; + +/** + * Save exceptions thrown while executing a job in a separate transaction. + * + * @see <a href="https://jira.jboss.org/jira/browse/JBPM-2691">JBPM-269... + * @author Alejandro Guizar + */ +public class JBPM2691Test extends AbstractDbTestCase { + + private long processDefinitionId; + + protected void setUp() throws Exception { + super.setUp(); + ProcessDefinition processDefinition = ProcessDefinition.parseXmlResource("org/jbpm/jbpm2691/processdefinition.xml"); + jbpmContext.deployProcessDefinition(processDefinition); + processDefinitionId = processDefinition.getId(); + + newTransaction(); + } + + protected void tearDown() throws Exception { + graphSession.deleteProcessDefinition(processDefinitionId); + super.tearDown(); + } + + public void testSaveExceptionInSeparateTx() { + ProcessInstance processInstance = jbpmContext.newProcessInstance("jobex"); + processInstance.signal(); + + processJobs(60 * 1000); + processInstance = jbpmContext.getProcessInstance(processInstance.getId()); + Token rootToken = processInstance.getRootToken(); + List jobs = jobSession.findJobsByToken(rootToken); + assertEquals(1, jobs.size()); + + Job job = (Job) jobs.get(0); + String exception = job.getException(); + assertNotNull(exception); + assert exception.indexOf("boom") != -1 : exception; + + // exception should leave job in its original state + assertEquals("async", rootToken.getNode().getName()); + } +} Property changes on: jbpm3/branches/jbpm-3.2-soa/modules/core/src/test/java/org/jbpm/jbpm2691/JBPM2691Test.java ___________________________________________________________________ Name: svn:keywords + Id Revision Name: svn:eol-style + native Modified: jbpm3/branches/jbpm-3.2-soa/modules/core/src/test/resources/org/jbpm/jbpm2574/processdefinition.xml =================================================================== --- jbpm3/branches/jbpm-3.2-soa/modules/core/src/test/resources/org/jbpm/jbpm2574/processdefinition.xml 2009-12-18 19:00:53 UTC (rev 5998) +++ jbpm3/branches/jbpm-3.2-soa/modules/core/src/test/resources/org/jbpm/jbpm2574/processdefinition.xml 2009-12-18 21:13:44 UTC (rev 5999) @@ -1,32 +1,23 @@ <?xml version="1.0" encoding="UTF-8"?> -<process-definition xmlns="" name="scripted"> +<process-definition name="scripted" xmlns="urn:jbpm.org:jpdl-3.2"> + <start-state name="start"> + <transition to="esb" /> + </start-state> - <start-state name="start"> - <transition to="esb"></transition> - </start-state> + <node name="esb"> + <action class="org.jbpm.mock.EsbActionHandler"> + <esbServiceName>paybills</esbServiceName> + <esbCategoryName>annoyances</esbCategoryName> + </action> + <transition to="end" /> + </node> + <end-state name="end" /> - <node name="esb"> - <action class="org.jbpm.mock.EsbActionHandler"> - <esbServiceName> - paybills - </esbServiceName> - <esbCategoryName> - annoyances - </esbCategoryName> - </action> - <transition to="end"></transition> - </node> + <event type="node-enter"> + <action class="org.jbpm.no.such.Class" /> + </event> - - <end-state name="end"></end-state> - - - <event type="node-enter"> - <action class="org.jbpm.no.such.Class"></action> - </event> - - </process-definition> \ No newline at end of file Added: jbpm3/branches/jbpm-3.2-soa/modules/core/src/test/resources/org/jbpm/jbpm2691/gpd.xml =================================================================== --- jbpm3/branches/jbpm-3.2-soa/modules/core/src/test/resources/org/jbpm/jbpm2691/gpd.xml (rev 0) +++ jbpm3/branches/jbpm-3.2-soa/modules/core/src/test/resources/org/jbpm/jbpm2691/gpd.xml 2009-12-18 21:13:44 UTC (rev 5999) @@ -0,0 +1,20 @@ +<?xml version="1.0" encoding="UTF-8"?> + +<root-container name="jobex" width="862" height="591"> + <node name="start" x="172" y="42" width="132" height="36"> + <edge> + <label x="5" y="-10"/> + </edge> + </node> + <node name="async" x="172" y="111" width="132" height="36"> + <edge> + <label x="5" y="-10"/> + </edge> + </node> + <node name="bomb" x="172" y="180" width="132" height="36"> + <edge> + <label x="5" y="-10"/> + </edge> + </node> + <node name="end" x="172" y="256" width="132" height="36"/> +</root-container> Property changes on: jbpm3/branches/jbpm-3.2-soa/modules/core/src/test/resources/org/jbpm/jbpm2691/gpd.xml ___________________________________________________________________ Name: svn:keywords + Id Revision Name: svn:eol-style + native Added: jbpm3/branches/jbpm-3.2-soa/modules/core/src/test/resources/org/jbpm/jbpm2691/processdefinition.xml =================================================================== --- jbpm3/branches/jbpm-3.2-soa/modules/core/src/test/resources/org/jbpm/jbpm2691/processdefinition.xml (rev 0) +++ jbpm3/branches/jbpm-3.2-soa/modules/core/src/test/resources/org/jbpm/jbpm2691/processdefinition.xml 2009-12-18 21:13:44 UTC (rev 5999) @@ -0,0 +1,18 @@ +<?xml version="1.0" encoding="UTF-8"?> + +<process-definition name="jobex" xmlns="urn:jbpm.org:jpdl-3.2"> + <start-state name="start"> + <transition to="async" /> + </start-state> + + <node name="async" async="true"> + <transition to="bomb" /> + </node> + + <node name="bomb"> + <action class="org.jbpm.jbpm2691.BombAction" /> + <transition to="end" /> + </node> + + <end-state name="end" /> +</process-definition> \ No newline at end of file Property changes on: jbpm3/branches/jbpm-3.2-soa/modules/core/src/test/resources/org/jbpm/jbpm2691/processdefinition.xml ___________________________________________________________________ Name: svn:keywords + Id Revision Name: svn:eol-style + native