[jbpm-commits] JBoss JBPM SVN: r5189 - in jbpm3/trunk/modules/core/src: test/java/org/jbpm and 1 other directories.

do-not-reply at jboss.org do-not-reply at jboss.org
Thu Jul 2 09:08:35 EDT 2009


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-2357</a>
+ * @author mputz at 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");
+    }
+  }
+}




More information about the jbpm-commits mailing list