3:31 p.m.
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&...]