Multi-instance sub process completion concurrency problem

As we transition to the new Hyland Community, some post links may be temporarily disrupted as we finalize redirects. We apologize for the inconvenience and are working to resolve this as quickly as possible. Details.

ochrons

I've got a multi-instance sub process defined which contains Java service tasks and timers. The timers are used for polling an external service (through a Java service task). If the results are not ready, the process will go back into the timer to wait.

Now, because of the timers, Activiti will use its internal thread pool to run the different executions in parallel in different threads. This is ok until the processes start getting completed at the same time. At completion, there are sometimes exceptions:

18-May-2012 20:37:35 org.activiti.engine.impl.interceptor.CommandContext close
SEVERE: Error while closing command context
org.activiti.engine.ActivitiOptimisticLockingException: VariableInstanceEntity[17] was updated by another transaction concurrently
   at org.activiti.engine.impl.db.DbSqlSession.flushUpdates(DbSqlSession.java:452)
‍‍‍‍‍‍

The variable (17) that it being updated is actually the nrOfActiveInstances counter for the sub process. This behavior seems quite silly to me, since it's nothing I can control directly.

I suppose I could work around this issue by introducing a Java Receive Task just before the sub process end event, to synchronize all Activiti controlled threads into my own thread.

Any real solutions to this problem?

- Otto

ochrons

So, I solved this with a workaround involving an additional Java Receive Task and a separate thread in my code that just signals that state whenever it's encountered.

Task definition in the process XML:
<receiveTask id="syncProcess" name="Sync"></receiveTask>
Code that creates a thread and checks for any process executions in that wait state and signals them:
<br>private RuntimeService runtimeService;<br>private Set<String> waitingProcesses = Collections.synchronizedSet(new HashSet<String>());<br>.<br>.<br>// create a thread for signaling waiting executions<br>Thread signalThread = new Thread(new Runnable() {<br>&nbsp;&nbsp;&nbsp; @Override<br>&nbsp;&nbsp;&nbsp; public void run() {<br>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; while (true) {<br>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; synchronized (waitingProcesses) {<br>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; Iterator<String> iterator = waitingProcesses.iterator();<br>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; while (iterator.hasNext()) {<br>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; String processId = iterator.next();<br>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; List<Execution> executions = runtimeService.createExecutionQuery().processInstanceId(processId)<br>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; .activityId("syncProcess").list();<br>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; for (Execution exec : executions) {<br>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; log.debug("Signaling waiting execution {}", exec.getId());<br>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; runtimeService.signal(exec.getId());<br>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; }<br>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; }<br>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; }<br>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; try {<br>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; Thread.sleep(1000);<br>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } catch (InterruptedException e) {<br>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; return;<br>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; }<br>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; }<br>&nbsp;&nbsp;&nbsp; }<br>});<br>signalThread.start();<br>
Would be great if there was a real solution to this problem instead of this kind of hack

- Otto