Update process variable from

[kaihuener]

Dear all,

I am struggling with updating a process variable from a java delegate that is called by an asynchronously executed multi-instance service task.
My setting:

• Simple process with a multi-instance service task

  st‍

  that is to be executed for a given collection of java beans and a following user task

  ut‍

  .

• st‍

  is marked as

  async="true"‍

  .
• A java delegate

  jd‍

  that does some work with a bean from the collection and
• A process variable

  progress‍

  that SHOULD be update by each

  jd‍

  call with the progress of processing the bean collection (i.e.

  nrOfCompletedInstances/nrOfInstances‍

  ) and that SHOULD be accessible from "outside" of

  jd‍

  execution.

My problem:
Setting

progress‍

within

jd‍

works fine by

int completed = (int) execution.getVariable("nrOfCompletedInstances") + 1;
int total = (int) execution.getVariable("nrOfInstances");
float progress = (float) completed / (float) total;
execution.setVariable("progress", progress)
‍‍‍‍‍‍

but when I read

progress‍

during process execution by

getProcessEngine().getRuntimeService().getVariable(processInstanceId, "progress");
‍‍‍

I always get the initial value and not the value that was set by

jd‍

.

Any hints what I am missing or how to "update" process variables from "inner" asynchronous executions?

Thank you in advance and best regards,
kai

[jbarrez]

That should be it. Are you sure the job executor is running for executing the async stuf.

If so, can you show your process xml and your java code so we can see what you're doing exactly.

[kaihuener]

Hi Joram,
Thank you for looking at this!
We do configuration in Java code:
<java>
// Create process engine and apply basic configuration
StandaloneProcessEngineConfiguration configuration = (StandaloneProcessEngineConfiguration) ProcessEngineConfiguration.createStandaloneProcessEngineConfiguration();
configuration.setProcessEngineName(PROCESS_ENGINE_NAME);

// Activate asynchronous job executor
configuration.setJobExecutorActivate(true);

// Further basic configuration
configuration.setDatabaseSchemaUpdate("true");
configuration.setHistory("none");

if (Configuration.MODE_ACTIVITI_TEST)
{ // Test mode
    logger.debug("Running in test mode ");
     // Load database configuration
     BasicDataSource dataSource = new BasicDataSource();
     dataSource.setDriverClassName("org.h2.Driver");
     dataSource.setUrl("jdbc:h2:mem:activiti;DB_CLOSE_DELAY=1000");
     configuration.setDataSource(dataSource);
}
else { /* Production mode … */ }

// Build process engine
configuration.buildProcessEngine();
</java>
I also tried the following configuration to use the new asynchronous job executor, same result:
<java>
configuration.setAsyncExecutorEnabled(true);
configuration.setAsyncExecutorActivate(true);
// configuration.setJobExecutorActivate(true);
</java>

The process is really simple:
<code>
<?xml version="1.0" encoding="UTF-8"?>
<definitions xmlns="http://www.omg.org/spec/BPMN/20100524/MODEL" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:activiti="http://activiti.org/bpmn" xmlns:bpmndi="http://www.omg.org/spec/BPMN/20100524/DI" xmlnsmgdc="http://www.omg.org/spec/DD/20100524/DC" xmlnsmgdi="http://www.omg.org/spec/DD/20100524/DI" typeLanguage="http://www.w3.org/2001/XMLSchema" expressionLanguage="http://www.w3.org/1999/XPath" targetNamespace="http://www.bei-sg.ch/cdl">
  <collaboration id="Collaboration">
    <participant id="poolCleansing" name="Cleansing" processRef="pCleansing"></participant>
  </collaboration>
  <process id="pCleansing" name="Cleansing" isExecutable="true">
    <laneSet id="laneSet_pCleansing">
      <lane id="laneUser" name="User">
        <flowNodeRef>eventStart</flowNodeRef>
        <flowNodeRef>endevent1</flowNodeRef>
        <flowNodeRef>exclusivegateway1</flowNodeRef>
        <flowNodeRef>utManageCleansing</flowNodeRef>
      </lane>
      <lane id="laneService" name="CDL Service">
        <flowNodeRef>stCleanse</flowNodeRef>
      </lane>
    </laneSet>
    <startEvent id="eventStart" name="Start" activiti:initiator="initiator">
      <extensionElements>
        <activiti:executionListener event="start" class="cdq.cdl.processes.delegates.CleansingStartListener"></activiti:executionListener>
      </extensionElements>
    </startEvent>
    <endEvent id="endevent1" name="End"></endEvent>
    <userTask id="utManageCleansing" name="Manage cleansing" activiti:assignee="${initiator}"></userTask>
    <serviceTask id="stCleanse" name="Cleanse address" activiti:async="true" activiti:exclusive="false" activiti:class="cdq.cdl.processes.delegates.CleanseAddress">
      <multiInstanceLoopCharacteristics isSequential="true" activiti:collection="${businessPartnerList}" activiti:elementVariable="businessPartner"></multiInstanceLoopCharacteristics>
    </serviceTask>
    <exclusiveGateway id="exclusivegateway1" name="Exclusive Gateway"></exclusiveGateway>
    <sequenceFlow id="flow2" sourceRef="utManageCleansing" targetRef="exclusivegateway1"></sequenceFlow>
    <sequenceFlow id="flow3" sourceRef="exclusivegateway1" targetRef="endevent1">
      <conditionExpression xsi:type="tFormalExpression"><![CDATA[${endProcess == true}]]></conditionExpression>
    </sequenceFlow>
    <sequenceFlow id="flow4" sourceRef="exclusivegateway1" targetRef="stCleanse"></sequenceFlow>
    <sequenceFlow id="flow5" sourceRef="stCleanse" targetRef="utManageCleansing"></sequenceFlow>
    <sequenceFlow id="flow6" sourceRef="eventStart" targetRef="utManageCleansing"></sequenceFlow>
  </process>
</definitions>
</code>
And the delegate code for the multi-instance task is unspectacular as well:
<java>
public class CleanseAddress implements JavaDelegate
{
    private final static Logger logger = LogManager.getLogger(CleanseAddress.class);

    @Override
    public void execute(DelegateExecution execution)
    {
        // Cleanse  data
        …

        // Calculate and report cleansing progress
        int completed = (int) execution.getVariable(Constants.ProcessVariables.NR_OF_COMPLETED_INSTANCES) + 1;
        int total = (int) execution.getVariable(Constants.ProcessVariables.NR_OF_INSTANCES);
        float progress = (float) completed / (float) total;
        execution.setVariable(Constants.ProcessVariables.PROGRESS, progress);
        logger.info(String.format("Cleansing progress: %.2f", progress));
    }
}
</java>

I really don't see the problem … The logger correctly reports the progress during process execution, but the process variable is 0.0 during the entire execution and becomes 1.0 after execution.

Thank you very much for any hint,
best,
kai

[trademak]

Hi Kai,

So you see the progress variable getting updated with every multi instance run?
But when the multi instance task has been completed, the same variable is the initial value?

Best regards,

[kaihuener]

Hi Tijs,
No, exactly the other way round
The progress is updates in the multi-instance run (i.e. the delegate, shown in the log), but the process variable (that is updated in the line before the log) holds the initial value (from before the multi-instance call) when I access it from outside the multi-instance execution. After the multi-instance execution, the process variable holds the last value that was set by the delegate, 1.0f. However, I wonder if this behavior is due to async execution? 
It seams to me that process variables are copied/cloned for an async execution and synchronized at next transaction commit, in my case after all multi-instance runs. Is that right?
What I am looking for is a "synchronized" process variable that can be written also from asynchronously executed activities. I now have implemented such a variable in a separate service class, but of course I would like to keep such information in the process …
Maybe you can briefly confirm that I understood the async behaviour (i.e. variable synchronization) correctly?

Thank you and best regards,
kai

[jbarrez]

I'm not really following here what is happening. So just to make sure, the variables you are fetched (nr of instances for example) are local to the execution itself. So yes, it could be that due to async execution you are seeing funny stuff here.  To solve that, you probably need to use process-instance scoped variables (instead of the execution local ones). However, that will lead to optimistic locking when going async of course.

[kaihuener]

Hi Joram,
Thank you for coming back. I am definitely using process-instance scoped variables, because they have the data I expect after the async and multi-instance stuff. However, I think I am on the wrong track in general – maybe +50k multi-instances, asynchronously executed without any user interaction, is not the intended use case for Activiti … I also encountered performance issues with persisting and loading "big data" process variables (+200 MB) and with cancellation of the async multi-instance executions. This may be due to insufficient Activiti knowledge, however,  I will go a classic way now with custom threads and a simple Activiti management process on top.
Thank you for your time,
best, kai

[jbarrez]

Those requirements do sound quite big. 200MB variables is something that will be very hard to do with Activiti (and databases i general).