[jBPM] - Concurrency problems with JBPM 5.4

Jon Kranes [https://community.jboss.org/people/jkranes] created the discussion

"Concurrency problems with JBPM 5.4"

To view the discussion, visit: https://community.jboss.org/message/828867#828867

--------------------------------------------------------------
I am working on a JBPM 5.4 / jBoss 7.1.1 application and have been unable to get it to run reliably using multiple threads.

The basic approach I am taking is as follows:

* Process instances are triggered by an incoming message (AMQP/ RabbitMQ) received by a message listener.  For optimal performance I need to be able to use multiple listener threads.   (Ultimately there will be multiple application server host instances as well but first things first).
* I am creating a new StatefulKnowlegeSession for each new ProcessInstance.
* I am using a local HumanTaskService (note that I am able to trigger concurrency problems mentioned below even while using a workflow that has no user tasks in it).

* I am disposing of the session and its associated handlers immediately after starting the process.  If I need to respond to an asynch signal (e.g. completion of a human task) I create the session by ID from the database (this is part of the scaling strategy - need to be able to continue the process at a future time on an entirely different machine).
* Mainly using Oracle (11g XE) but have also tested with MySQL and see the exact same issues.


I am using Spring with declarative transactions.  I am using JTA transactions through the jBoss platform JTA transaction manager.

Here's what I am experiencing:

* Everything works fine as long as as I run a single thread only (i.e. have only one instance of my message listener class).  I can run this way for hours without a single exception being thrown.
* With two or more message listeners, I get exceptions (not necessarily right away, but if I run enough iterations it happens pretty quickly).
* I see a variety of different exceptions each time I run the program.  The exceptions vary but all are related to inability to commit, invalid sessions, inability to find a certain piece of data (the session or process instance), deadlocks, concurrent modification exceptions, etc.  All the kinds of things you would expect in non threadsafe code.
* Once an exception is thrown on a given thread, it seems that the thread will just keep throwing the 'Entity Manager is Closed' exception.  I am guessing there is something happenging here with a ThreadLocal tied to the thread that never gets cleared up once problems occur on that thread.  If I run long enough, all threads end up with this problem and the system basically becomes useless and needs a full restart.


So without trying to debug each and every exception one at a time, I'm wondering if there is anything obvious I am doing in my code that would lead to a non threadsafe condition.  I have read several threads on issues surrounding the StatefulKnowledgeSession so I have deliberately chosen to NOT share sessions between threads for that reason (and others).

A few main questions also:

* Wrapping the Spring JTA PlatformTransactionManager in the DroolsSpringTransactionManager: i have seen this in several examples but have not seen any good explanation of whether/when it is necessary.
* Use of the DroolsSpringJPAManager: again, have seen this in examples but no real documentation explaining it.
* Should the StatefulKnowledgeSession be created inside a transaction?  Must it be the same transaction that it is used in?   I know that it has to be disposed outside of the transaction boundaries.



Thanks for any help or suggestions.  I'm happy to provide more code as needed; I wanted to keep this limited to what seems relevant rather than zip up an entire complex application and force people to wade through all of it.

-- Jon Kranes



Relevant code:

I have a wrapper class to hold a reference to the StatefulKnowledgeSession and its associated handlers, to make it easier to manage disposal:


public class SessionWrapper {
  
 private static Logger log = Logger.getLogger(SessionWrapper.class);
  
 private StatefulKnowledgeSession session;
 private GenericHTWorkItemHandler htHandler;
 private JPAWorkingMemoryDbLogger dbLogger;
 private LocalTaskService localTaskService;
  
 public void dispose() {
 try {
                                                             log.info("disposing local task service");
 localTaskService.dispose();
 }
 catch (Exception e) {
                                                             log.error(e);
 }
 try {
                                                             // log.info("disposing db logger");
                                                             // this does not do anything.
 // dbLogger.dispose();
 }
 catch (Exception e) {
                                                             log.error(e);
 }
 try {
                                                             log.info("disposing ht handler");
                                                             htHandler.dispose();
 }
 catch (Exception e) {
                                                             log.error(e);
 }
 try {
                                                             log.info("Disposing session ID " + session.getId());
                                                             session.dispose();
 }
 catch (Exception e) {
 log.error(e);
 }
 }
  
  
 public StatefulKnowledgeSession getSession() {
 return session;
 }
 public void setSession(StatefulKnowledgeSession session) {
 this.session = session;
 }
 public GenericHTWorkItemHandler getHtHandler() {
 return htHandler;
 }
 public void setHtHandler(GenericHTWorkItemHandler htHandler) {
 this.htHandler = htHandler;
 }
 public JPAWorkingMemoryDbLogger getDbLogger() {
 return dbLogger;
 }
 public void setDbLogger(JPAWorkingMemoryDbLogger dbLogger) {
 this.dbLogger = dbLogger;
 }
 public LocalTaskService getLocalTaskService() {
 return localTaskService;
 }
 public void setLocalTaskService(LocalTaskService localTaskService) {
 this.localTaskService = localTaskService;
 }
 
}
 


The SessionWrapper instance is created here


public class KnowledgeSessionFactory {
  
 @Autowired
 KnowledgeBase kbase;
  
 @Autowired
 org.jbpm.task.service.TaskService taskService;
  
 @Autowired
 EntityManagerFactory emf;
  
 @Autowired
 EventDrivenTaskHandler handler;
  
 @Autowired
         DroolsSpringTransactionManager txManager;
    
    @Autowired
    AuditService auditService;
    
    @Autowired
    AuditTaskHandler auditHandler;
   
    private static final Logger log = Logger.getLogger(KnowledgeSessionFactory.class);
    
 public SessionWrapper createKnowledgeSession() {
 log.info("Creating new StatefulKnowledgeSession");
 StatefulKnowledgeSession ksession = null;
 try {
                                                             ksession = JPAKnowledgeService.newStatefulKnowledgeSession( kbase, null, getEnvironment() );
 }
 catch (Exception e) {
                                                             log.error("Unable to create sesssion!");
                                                             log.error(e);
                                                             throw new RuntimeException(e);
 }
 log.info("Session ID: " + ksession.getId());
 return createSessionWrapper(ksession); 
 }
  
 public SessionWrapper createKnowledgeSession(int sessionId) {
 log.info("Loading StatefulKnowledgeSession for ID: " + sessionId);
 StatefulKnowledgeSession ksession = JPAKnowledgeService.loadStatefulKnowledgeSession(sessionId, kbase, null, getEnvironment());
 return createSessionWrapper(ksession); 
 } 
  
 private SessionWrapper createSessionWrapper(StatefulKnowledgeSession ksession) {
 LocalTaskService taskClient = new LocalTaskService(taskService);
 GenericHTWorkItemHandler htHandler = new LocalHTWorkItemHandler(taskClient, ksession);
  
 ksession.getWorkItemManager().registerWorkItemHandler("Human Task", htHandler);
 ksession.getWorkItemManager().registerWorkItemHandler("Task", handler);
 ksession.getWorkItemManager().registerWorkItemHandler("Audit", auditHandler);
 ksession.addEventListener(new DroolsListener());
  
 // This allows access of the global in script tasks
 ksession.setGlobal("auditService", auditService);
  
 JPAWorkingMemoryDbLogger dbLogger = new JPAWorkingMemoryDbLogger(ksession);
 SessionWrapper wrapper = new SessionWrapper();
 wrapper.setSession(ksession);
 wrapper.setHtHandler(htHandler);
 wrapper.setDbLogger(dbLogger);
 wrapper.setLocalTaskService(taskClient);
 return wrapper; 
               }
  
 private Environment getEnvironment() {
 log.info("creating environment");
 env = KnowledgeBaseFactory.newEnvironment();
 env.set(EnvironmentName.ENTITY_MANAGER_FACTORY, emf );
 env.set(EnvironmentName.TRANSACTION_MANAGER, txManager);
 PersistenceContextManager persistenceContextManager = new DroolsSpringJpaManager(env);
 env.set(EnvironmentName.PERSISTENCE_CONTEXT_MANAGER, persistenceContextManager);
 log.info("environment created"); 
 return env;
 }
 
}
 
 



Here is the message driven bean that initiates starting new process instances:

public class StartProcessListener implements MessageListener {
  
 private static Logger log = Logger.getLogger(StartProcessListener.class);
  
 @Autowired
 EventService service;
  
 @Autowired
 KnowledgeSessionFactory factory;
  
 @Autowired
 SerializerMessageConverter converter;
  
 public void onMessage(Message message) {
 SessionWrapper wrapper = null;
 try {
                                                             Map<String, Object> map = (Map<String, Object>)converter.fromMessage(message);
                                                             Map<String, Object> params = new HashMap<String, Object>();
                                                             params.put("returnStatus", "VALID");
                                                             params.put("return", map.get("return"));
  
                                                             String processId = (String)map.get("processId");
                                                             try {
                                                                            wrapper = service.createSessionWrapper();
                                                                            service.startProcess(wrapper, processId, params);
                                                             }
                                                             catch (Exception e) {
                                                                            log.error(e);
                                                             }
 }
 catch (Exception e) {
                                                             log.error(e);
 }
 finally {
                                                             if (wrapper != null) {
                                                                            wrapper.dispose(); 
                                                             }
 }
 }
 



And finally the service class that uses declarative transactions:


public class EventServiceImpl implements EventService {
  
 private static Logger log = Logger.getLogger(EventServiceImpl.class);
  
 @Autowired
 KnowledgeSessionFactory factory;
  
 @Autowired
 TaskService taskService;
  
 @Override
 @Transactional
 public SessionWrapper createSessionWrapper() {
 return factory.createKnowledgeSession();
 }
  
 @Transactional
 @Override
 public void startProcess(SessionWrapper wrapper, String processId, Map<String, Object> params) {
 // SessionWrapper wrapper = factory.createKnowledgeSession();
 StatefulKnowledgeSession ksession = wrapper.getSession();
  
 // initializes the human task handler for the session
 // org.jbpm.task.TaskService lts = LocalHumanTaskService.getTaskService(ksession);
  
 // this allows us to insert the processInstance into the session before starting it.
 ProcessInstance processInstance = ksession.createProcessInstance(processId, params); 
 // for drools
 ksession.insert(processInstance);
  
 log.info("Starting process instance");
 ksession.startProcessInstance(processInstance.getId());
 //return wrapper;
 }
  
 @Transactional
 @Override
               public SessionWrapper completeWorkItem(int sessionId, long workItemId, Map<String, Object> results) {
 SessionWrapper wrapper = factory.createKnowledgeSession(sessionId);
 StatefulKnowledgeSession ksession = wrapper.getSession();
 completeWorkItem(ksession, workItemId, results); 
 return wrapper;
 }
  
 @Override
 @Transactional
 public SessionWrapper completeTask(int sessionId, Long taskId, Map<String, Object> results) {
 log.info("Completing task: " + taskId);
  
 for (String key : results.keySet()) {
                                                             log.info(key + "=" + results.get(key));
 }
  
 // LocalTaskService lts = new LocalTaskService(taskService);
 // StatefulKnowledgeSession ksession = factory.getKnowledgeSession(sessionId);
 SessionWrapper wrapper = factory.createKnowledgeSession(sessionId);
 StatefulKnowledgeSession ksession = wrapper.getSession();
  
 // org.jbpm.task.TaskService lts = LocalHumanTaskService.getTaskService(ksession);
 org.jbpm.task.TaskService lts = wrapper.getLocalTaskService();
 Task task = lts.getTask(taskId);
 // int sessionId = task.getTaskData().getProcessSessionId();
 // completeWorkItem(sessionId, task.getTaskData().getWorkItemId(), results);
  
 try {
                                                             lts.start(taskId, "USER");
 }
 catch (Exception e) {
                                                             log.error(e);
 }
  
 try {
                                                             ContentData contentData = ContentMarshallerHelper.marshal(results, null);
                                                             lts.complete(taskId, "USER", contentData);
                                                             lts.disconnect(); // does this do anything useful?
 }
 catch (Exception e) {
                                                             log.error(e);
 }
  
 // for drools
 ProcessInstance pi = ksession.getProcessInstance(task.getTaskData().getProcessInstanceId());
 ksession.insert(pi);
  
 completeWorkItem(ksession, task.getTaskData().getWorkItemId(), results);
  
 return wrapper;
  
 }
  
 private void completeWorkItem(StatefulKnowledgeSession ksession, Long workItemId, Map<String, Object> results) {
 log.info("Completing work item: " + workItemId);
 ksession.getWorkItemManager().completeWorkItem(workItemId, results); 
 }
 
 
}
 




Spring configuration:


<?xml version="1.0" encoding="UTF-8"?>


<beans ... >
                 
    <context:annotation-config />
    
    <!-- Loads all classes annotated with @Component or @Service -->
    <context:component-scan base-package="org.mitre.irs.eda.bpm"/>
   
    <context:property-placeholder location="classpath:eda.properties" order="2"/>     
    
          <jee:jndi-lookup id="emf" jndi-name="persistence/emf"/>
  
          <!-- Spring JtaTransactionManager -->
          <bean id="jtaTxManager" class="org.springframework.transaction.jta.JtaTransactionManager" />
  
          <bean id="txManager" class="org.drools.container.spring.beans.persistence.DroolsSpringTransactionManager"
                    c:ptm-ref="jtaTxManager" />


          <tx:annotation-driven transaction-manager="jtaTxManager" /> 
  
          <jpa:repositories base-package="org.mitre.irs.eda.common.repo" entity-manager-factory-ref="emf" transaction-manager-ref="jtaTxManager" />
  
          <bean id="jpaDialect" class="org.springframework.orm.jpa.vendor.HibernateJpaDialect" />  
  
          <!-- JBPM -->
          <bean id="sel" class="org.drools.SystemEventListenerFactory"
                    factory-method="getSystemEventListener" />

          <bean id="jbpmTaskService" class="org.jbpm.task.service.TaskService"
                    c:emf-ref="emf" c:systemEventListener-ref="sel" /> 

         <bean id="kbaseFactory" class="org.mitre.irs.eda.bpm.jbpm.KnowledgeBaseFactory" />
    
         <bean id="kbase" factory-bean="kbaseFactory" factory-method="newKnowledgeBase" />

            
    
          <!-- Rabbit MQ -->
          <rabbit:connection-factory id="connectionFactory" channel-cache-size="20"
                    username="${amqp.username}" password="${amqp.password}" host="${amqp.host}" port="${amqp.port}" />
  
          <rabbit:admin id="admin" connection-factory="connectionFactory" />
          <bean id="serializerConverter" class="org.springframework.amqp.support.converter.SerializerMessageConverter" />
          <rabbit:template id="amqpTemplate" connection-factory="connectionFactory" message-converter="serializerConverter"/>
          <rabbit:queue id="start" name="process.start" />
          <rabbit:queue id="task.start" name="process.task.start" />
          <rabbit:queue id="task.complete" name="process.task.complete" />
          <bean id="startListener" class="org.mitre.irs.eda.bpm.listener.StartProcessListener" />
          <bean id="taskCompleteListener" class="org.mitre.irs.eda.bpm.listener.TaskCompleteListener" />
          <rabbit:listener-container connection-factory="connectionFactory" concurrency="4">
                    <rabbit:listener queues="start" ref="startListener" />
                    <rabbit:listener queues="task.complete" ref="taskCompleteListener" />
          </rabbit:listener-container>   
    

          
</beans>
--------------------------------------------------------------

Reply to this message by going to Community
[https://community.jboss.org/message/828867#828867]

Start a new discussion in jBPM at Community
[https://community.jboss.org/choose-container!input.jspa?contentType=1&containerType=14&container=2034]

_______________________________________________
jboss-user mailing list
[email protected]
https://lists.jboss.org/mailman/listinfo/jboss-user