8:08 a.m.
Author: mputz
Date: 2009-07-02 09:08:35 -0400 (Thu, 02 Jul 2009)
New Revision: 5189
Added:
jbpm3/trunk/modules/core/src/test/java/org/jbpm/jbpm2375/
jbpm3/trunk/modules/core/src/test/java/org/jbpm/jbpm2375/JBPM2375Test.java
Modified:
jbpm3/trunk/modules/core/src/main/java/org/jbpm/job/executor/JobExecutor.java
jbpm3/trunk/modules/core/src/main/java/org/jbpm/job/executor/LockMonitorThread.java
Log:
Fix for JBPM-2375 - JobExecutorThread is terminated after an Error: LockMonitorThread
restarts terminated JobExecutorThreads.
Modified: jbpm3/trunk/modules/core/src/main/java/org/jbpm/job/executor/JobExecutor.java
===================================================================
---
jbpm3/trunk/modules/core/src/main/java/org/jbpm/job/executor/JobExecutor.java 2009-07-02
13:03:14 UTC (rev 5188)
+++
jbpm3/trunk/modules/core/src/main/java/org/jbpm/job/executor/JobExecutor.java 2009-07-02
13:08:35 UTC (rev 5189)
@@ -8,6 +8,7 @@
import java.util.HashMap;
import java.util.HashSet;
import java.util.Hashtable;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -50,7 +51,8 @@
for (int i = 0; i < nbrOfThreads; i++) {
startThread();
}
- lockMonitorThread = new LockMonitorThread(this);
+ lockMonitorThread = new LockMonitorThread(getLockMonitorThreadName(), this);
+ lockMonitorThread.start();
isStarted = true;
}
else {
@@ -102,9 +104,31 @@
if (lockMonitorThread != null) lockMonitorThread.join();
}
+
+ public void ensureThreadsAreActive() {
+ List<String> deadThreads = new ArrayList<String>();
+ for (Iterator i = threads.values().iterator(); i.hasNext();) {
+ Thread thread = (Thread) i.next();
+ if (!thread.isAlive()) {
+ if(log.isDebugEnabled())
+ {
+ log.debug("detected dead thread '" + thread.getName() +
"'");
+ }
+ deadThreads.add(thread.getName());
+ i.remove();
+ }
+ }
+ for (int i = 0; i<deadThreads.size(); i++) {
+ startThread(deadThreads.get(i));
+ }
+ }
- protected synchronized void startThread() {
- String threadName = getNextThreadName();
+ protected void startThread() {
+ startThread(getNextThreadName());
+ }
+
+ protected synchronized void startThread(String name) {
+ String threadName = name;
Thread thread = createThread(threadName);
threads.put(threadName, thread);
if(log.isDebugEnabled())
@@ -129,6 +153,10 @@
private String getThreadName(int index) {
return name + '@' + getHostAddress() + ':' + index;
}
+
+ private String getLockMonitorThreadName() {
+ return name + ':' + LockMonitorThread.DEFAULT_NAME + '@' +
getHostAddress();
+ }
private static String getHostAddress() {
if (hostName == null) {
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-07-02
13:03:14 UTC (rev 5188)
+++
jbpm3/trunk/modules/core/src/main/java/org/jbpm/job/executor/LockMonitorThread.java 2009-07-02
13:08:35 UTC (rev 5189)
@@ -15,6 +15,8 @@
public class LockMonitorThread extends Thread {
+ public static final String DEFAULT_NAME = "LockMonitorThread";
+
JbpmConfiguration jbpmConfiguration;
int lockMonitorInterval;
int maxLockTime;
@@ -23,6 +25,11 @@
volatile boolean isActive = true;
public LockMonitorThread(JobExecutor jobExecutor) {
+ this(DEFAULT_NAME, jobExecutor);
+ }
+
+ public LockMonitorThread(String name, JobExecutor jobExecutor) {
+ super(name);
jbpmConfiguration = jobExecutor.getJbpmConfiguration();
lockMonitorInterval = jobExecutor.getLockMonitorInterval();
maxLockTime = jobExecutor.getMaxLockTime();
@@ -44,6 +51,7 @@
while (isActive) {
try {
unlockOverdueJobs();
+ jbpmConfiguration.getJobExecutor().ensureThreadsAreActive();
if (isActive && lockMonitorInterval > 0) {
sleep(lockMonitorInterval);
}
Added: jbpm3/trunk/modules/core/src/test/java/org/jbpm/jbpm2375/JBPM2375Test.java
===================================================================
--- jbpm3/trunk/modules/core/src/test/java/org/jbpm/jbpm2375/JBPM2375Test.java
(rev 0)
+++ jbpm3/trunk/modules/core/src/test/java/org/jbpm/jbpm2375/JBPM2375Test.java 2009-07-02
13:08:35 UTC (rev 5189)
@@ -0,0 +1,123 @@
+package org.jbpm.jbpm2375;
+
+import org.jbpm.JbpmConfiguration;
+import org.jbpm.db.AbstractDbTestCase;
+import org.jbpm.graph.def.ActionHandler;
+import org.jbpm.graph.def.Event;
+import org.jbpm.graph.def.EventCallback;
+import org.jbpm.graph.def.ProcessDefinition;
+import org.jbpm.graph.exe.ExecutionContext;
+import org.jbpm.graph.exe.ProcessInstance;
+import org.jbpm.job.executor.JobExecutor;
+
+/**
+ * Test if the JobExecutorThread recovers from an Error
+ *
+ * @see <a
href="https://jira.jboss.org/jira/browse/JBPM-2357">JBPM-235...
+ * @author mputz(a)redhat.com
+ * @since 30-Jun-2009
+ */
+public class JBPM2375Test extends AbstractDbTestCase {
+
+ private static final int TEST_TIMEOUT = 10 * 1000;
+
+ private JobExecutor jobExecutor = new JobExecutor();
+ private long processDefinitionId;
+
+ // a process definition with two timers moving the token forward
+ // the second state has an action associated with the node-enter event,
+ // which can simulate an Error condition by throwing a NoClassDefFoundError
+ private static final String PROCESS_DEFINITION = "<process-definition
name='jbpm2375-timer-error-test'>"
+ + " <event type='process-end'>"
+ + " <action expression='#{eventCallback.processEnd}'
/>"
+ + " </event>"
+ + " <start-state name='start'>"
+ + " <transition to='state1'
name='to_state1'/>"
+ + " </start-state>"
+ + " <state name='state1'>"
+ + " <timer name='moveToDefaultEndAfter1second' duedate='1
second' transition='to_state2'/>"
+ + " <transition to='state2'
name='to_state2'/>"
+ + " </state>"
+ + " <state name='state2'>"
+ + " <timer name='moveToEndAfter1second' duedate='1
second' transition='to_end'/>"
+ + " <event type='node-enter'>"
+ + " <action name='exceptionTest' class='"
+ + TimerExceptionAction.class.getName()
+ + "'>"
+ + " </action>"
+ + " </event>"
+ + " <transition to='end' name='to_end'/>"
+ + " </state>"
+ + " <end-state name='end' />"
+ + "</process-definition>";
+
+
+ @Override
+ protected void setUp() throws Exception {
+ super.setUp();
+
+ ProcessDefinition processDefinition =
ProcessDefinition.parseXmlString(PROCESS_DEFINITION);
+ jbpmContext.deployProcessDefinition(processDefinition);
+ newTransaction();
+ processDefinitionId = processDefinition.getId();
+
+ getJbpmConfiguration().getJobExecutor().setLockMonitorInterval(TEST_TIMEOUT/2);
+ startJobExecutor();
+
+
+ }
+
+ @Override
+ protected void tearDown() throws Exception {
+ stopJobExecutor();
+
+ graphSession.deleteProcessDefinition(processDefinitionId);
+
+ EventCallback.clear();
+ super.tearDown();
+ }
+
+ // check if the process ends correctly if no Error is thrown
+ public void testTimerWithoutErrorAction() {
+ runTimerErrorAction(Boolean.FALSE);
+ }
+
+ // check if the process ends correctly if an Error is thrown in the ActionHandler
+ public void testTimerWithErrorAction() {
+ runTimerErrorAction(Boolean.TRUE);
+ }
+
+ private void runTimerErrorAction(Boolean withError) {
+
+ // kick off process instance
+ ProcessDefinition processDefinition =
graphSession.loadProcessDefinition(processDefinitionId);
+ ProcessInstance processInstance = new ProcessInstance(processDefinition);
+ processInstance.getContextInstance().setVariable("eventCallback", new
EventCallback());
+ processInstance.getContextInstance().setVariable("throwError", withError);
+ processInstance.signal();
+ jbpmContext.save(processInstance);
+
+ commitAndCloseSession();
+ try {
+ EventCallback.waitForEvent(Event.EVENTTYPE_PROCESS_END, TEST_TIMEOUT);
+
+ waitForJobs(TEST_TIMEOUT);
+
+ } finally {
+ beginSessionTransaction();
+ }
+
+ }
+
+
+ public static class TimerExceptionAction implements ActionHandler {
+
+ private static final long serialVersionUID = 1L;
+
+ public void execute(ExecutionContext executionContext) throws Exception {
+ Boolean throwError =
(Boolean)executionContext.getVariable("throwError");
+ if (throwError)
+ throw new NoClassDefFoundError("org.jbpm.no.class.Class");
+ }
+ }
+}