ExeuctionQuery filtering with 'messageEventSubscription' and 'variableValueEquals' does not seem to work

[rallen1]

Hello,

I am having some trouble finding an execution listening for a message and filtering on a specific variable.

It is possible I am using the API incorrectly, but there seems to be an issue when I attempt to query executions for a specific messageEventSubscription and variableValueEquals.

      // Create a basic query on the intermediate message. 
      ExecutionQuery query = runtimeService.createExecutionQuery()
            .processInstanceId(processInstance.getProcessInstanceId())
            .messageEventSubscriptionName(MESSAGE);

      // Prove a query is returned.
      Execution execution = query.singleResult();
      assertNotNull(execution);
      // Prove our global variable is in the execution and its value is what is expected.
      assertEquals(runtimeService.getVariable(execution.getId(), KEY), VALUE);
      
      // If more than one query were returned above, we should be able to refine further
      // using variableValuesEqual. For some reason variableValueEquals and ignoreCase is failing.
      Execution refinedExecution = query.variableValueEquals(KEY, VALUE).singleResult();
      // !!! Fails. !!!
      assertNotNull(refinedExecution);
‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍

I wrote a unit test demonstrating the issue with the unit test template. I wasn't able to register with jira (codehaus had a random server error when I verified my email), so I apologize if this is the wrong place to post this.

My pom.xml is using the versions that reflect my actual project:

      <dependency>
         <groupId>org.activiti</groupId>
         <artifactId>activiti-engine</artifactId>
         <version>5.17.0</version>
      </dependency>
      <dependency>
         <groupId>com.h2database</groupId>
         <artifactId>h2</artifactId>
         <version>1.4.181</version>
         <scope>test</scope>
      </dependency>
‍‍‍‍‍‍‍‍‍‍‍‍‍

I tried switching the H2 version to 1.3.168, but it did not seem to make a difference.

I attached the unit test as well as the only two files I changed from the template (besides the pom.xml).

Any help is greatly appreciated.

Thank you.

[vasile_dirla]

Confirmed,
This happens when a new execution is created because the variables remains linked to the original process execution and then during the query by variables the new execution instance has no relation to them so there will be no execution to return)

AtomicOperationTransitionCreateScope.java
method: public void execute(InterpretableExecution execution).

I think this is a bug, I'll focus on it and find a solution.

[jbarrez]

> I think this is a bug, I'll focus on it and find a solution.

This is not a bug. Variables are stored on the process instance level. Your query won't return the results cause there is no execution that is 1) waiting at the message event and 2) has the variables. If you check the db, you'll see there are two executions.

Now, I do agree it would be a nice *feature* to allow this in the query api.

If this is crucial, a quick workaround would be to set the variables to the local execution (eg using an execution listener)

[rallen1]

I am not sure I understand adding an execution listener, but I would prefer to avoid that approach as it would make defining the bpmns I have in mind more difficult.

After working on the question here:

http://forums.activiti.org/content/signaling-event-process-instance-and-all-its-sub-processes

I came up with another unit test and bpmn that I think really illustrates the issue I am having. Perhaps my BPMN pattern is not intelligently thought out, but the idea is to be able to have Activiti capable of accepting intermediate messages that are within subprocesses, specifically multi instance subprocesses.

MyUnitTest.java

<java>
package org.activiti;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.activiti.engine.RuntimeService;
import org.activiti.engine.runtime.Execution;
import org.activiti.engine.runtime.ExecutionQuery;
import org.activiti.engine.runtime.ProcessInstance;
import org.activiti.engine.test.ActivitiRule;
import org.activiti.engine.test.Deployment;
import org.junit.Rule;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class MyUnitTest {
private static final Logger LOG = LoggerFactory.getLogger( MyUnitTest.class );

private static final String ITEMS = "items";
private static final String SUBKEY = "item";
private static final String ITEM1 = "AAAAA";
private static final String ITEM2 = "BBBBB";
private static final String MESSAGE = "waitingMessage";

@Rule
public ActivitiRule activitiRule = new ActivitiRule();

@Test
@Deployment(resources = {"org/activiti/test/my-process.bpmn20.xml"})
public void test_eventSubscription_FilterByVariable() {
 
  RuntimeService runtimeService = activitiRule.getRuntimeService();

  // Inject 'items' for the subprocess to spawn multiple instances with.
  Map<String, Object> variables = new HashMap<String, Object>();
  List<String> items = new ArrayList<String>(2);
  items.add(ITEM1);
  items.add(ITEM2);
  variables.put(ITEMS, items);
 
  ProcessInstance processInstance = runtimeService.startProcessInstanceByKey("my-process", variables);
  assertNotNull(processInstance);
 
  // Prove we have two subprocesses subscribed to the intermediate message.
  ExecutionQuery query = runtimeService.createExecutionQuery()
    .processInstanceId(processInstance.getProcessInstanceId())
    .messageEventSubscriptionName(MESSAGE);
 
  assertTrue(query.list().size() == 2);
 
  // Can we message one of the subprocesses?
  // We know each subprocess should have a unique variable value for "item".
  ExecutionQuery item1Query = runtimeService.createExecutionQuery()
    .variableValueEquals(SUBKEY, ITEM1);
  // We found the "parent"/ ScopedExecution with the variables for this subprocess.
  assertTrue(item1Query.list().size() == 1);
 
  Execution scopedExecution = item1Query.singleResult();
 
  // The problem is how do we uniquely identify the subscribed execution with our ScopedExecution?
  // The process instance ID of the scoped execution is the same as the main process instance.
  assertTrue(scopedExecution.getProcessInstanceId().equals(processInstance.getProcessInstanceId()));
 
  // Is there a piece of information can be used to link the ScopedExecution and subprocess subscribed execution?
  // We could solve this by looping through each subscribed execution and checking
  //
  // if (runtimeService.getVariable(subscribedExecutionId, SUBKEY).equals(ITEM1))
  //
  // But this seems like it would scale poorly as the number of items grows. In my case, items could be hundreds.
 
  // Send a message to the ITEM1 execution somehow.
 
  // Prove only one subprocess is still waiting.
  ExecutionQuery finalQuery = runtimeService.createExecutionQuery()
    .processInstanceId(processInstance.getProcessInstanceId())
    .messageEventSubscriptionName(MESSAGE);
 
  assertTrue(finalQuery.list().size() == 1);
 
}
}
</java>

my-process.bpmn20.xml

<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.activiti.org/test">
  <message id="waitingMessage" name="waitingMessage"></message>
  <process id="my-process" isExecutable="true">
    <startEvent id="startevent1" name="Start"></startEvent>
    <subProcess id="subprocess1" name="For each item in items">
      <multiInstanceLoopCharacteristics isSequential="false" activiti:collection="items" activiti:elementVariable="item"></multiInstanceLoopCharacteristics>
      <startEvent id="startevent2" name="Start"></startEvent>
      <intermediateCatchEvent id="messageintermediatecatchevent1" name="MessageCatchEvent">
        <messageEventDefinition messageRef="waitingMessage"></messageEventDefinition>
      </intermediateCatchEvent>
      <endEvent id="endevent2" name="End"></endEvent>
      <sequenceFlow id="flow3" name="waitingMessage" sourceRef="messageintermediatecatchevent1" targetRef="endevent2"></sequenceFlow>
      <sequenceFlow id="flow4" sourceRef="startevent2" targetRef="messageintermediatecatchevent1"></sequenceFlow>
    </subProcess>
    <endEvent id="endevent1" name="End"></endEvent>
    <sequenceFlow id="flow1" sourceRef="subprocess1" targetRef="endevent1"></sequenceFlow>
    <sequenceFlow id="flow2" sourceRef="startevent1" targetRef="subprocess1"></sequenceFlow>
  </process>
  <bpmndi:BPMNDiagram id="BPMNDiagram_my-process">
    <bpmndi:BPMNPlane bpmnElement="my-process" id="BPMNPlane_my-process">
      <bpmndi:BPMNShape bpmnElement="startevent1" id="BPMNShape_startevent1">
        <omgdc:Bounds height="35.0" width="35.0" x="210.0" y="240.0"></omgdc:Bounds>
      </bpmndi:BPMNShape>
      <bpmndi:BPMNShape bpmnElement="subprocess1" id="BPMNShape_subprocess1">
        <omgdc:Bounds height="205.0" width="321.0" x="320.0" y="155.0"></omgdc:Bounds>
      </bpmndi:BPMNShape>
      <bpmndi:BPMNShape bpmnElement="startevent2" id="BPMNShape_startevent2">
        <omgdc:Bounds height="35.0" width="35.0" x="350.0" y="240.0"></omgdc:Bounds>
      </bpmndi:BPMNShape>
      <bpmndi:BPMNShape bpmnElement="messageintermediatecatchevent1" id="BPMNShape_messageintermediatecatchevent1">
        <omgdc:Bounds height="35.0" width="35.0" x="460.0" y="240.0"></omgdc:Bounds>
      </bpmndi:BPMNShape>
      <bpmndi:BPMNShape bpmnElement="endevent2" id="BPMNShape_endevent2">
        <omgdc:Bounds height="35.0" width="35.0" x="540.0" y="240.0"></omgdc:Bounds>
      </bpmndi:BPMNShape>
      <bpmndi:BPMNShape bpmnElement="endevent1" id="BPMNShape_endevent1">
        <omgdc:Bounds height="35.0" width="35.0" x="686.0" y="240.0"></omgdc:Bounds>
      </bpmndi:BPMNShape>
      <bpmndi:BPMNEdge bpmnElement="flow3" id="BPMNEdge_flow3">
        <omgdi:waypoint x="495.0" y="257.0"></omgdi:waypoint>
        <omgdi:waypoint x="540.0" y="257.0"></omgdi:waypoint>
        <bpmndi:BPMNLabel>
          <omgdc:Bounds height="14.0" width="100.0" x="748.0" y="428.0"></omgdc:Bounds>
        </bpmndi:BPMNLabel>
      </bpmndi:BPMNEdge>
      <bpmndi:BPMNEdge bpmnElement="flow4" id="BPMNEdge_flow4">
        <omgdi:waypoint x="385.0" y="257.0"></omgdi:waypoint>
        <omgdi:waypoint x="460.0" y="257.0"></omgdi:waypoint>
      </bpmndi:BPMNEdge>
      <bpmndi:BPMNEdge bpmnElement="flow1" id="BPMNEdge_flow1">
        <omgdi:waypoint x="641.0" y="257.0"></omgdi:waypoint>
        <omgdi:waypoint x="686.0" y="257.0"></omgdi:waypoint>
      </bpmndi:BPMNEdge>
      <bpmndi:BPMNEdge bpmnElement="flow2" id="BPMNEdge_flow2">
        <omgdi:waypoint x="245.0" y="257.0"></omgdi:waypoint>
        <omgdi:waypoint x="320.0" y="257.0"></omgdi:waypoint>
      </bpmndi:BPMNEdge>
    </bpmndi:BPMNPlane>
  </bpmndi:BPMNDiagram>
</definitions>
</code>

Any help or thoughts are appreciated. Thanks.

[jbarrez]

> I am not sure I understand adding an execution listener

An execution listener would simply make sure the global variable is available on the local scope.

I do understand your problem you are having. I'm just saying that's the way it is in Activiti: variables are coupled with a process instance, not an execution. Unless you explicitly define them as execution-local.

[rallen1]

As an update:

I resolved this using the execution listener mechanism as suggested by Joram. For whomever may read this in the future, just use this inside your execution listener:

<java>
runtimeService.setVariablesLocal(executionId, variables);
</java>

And then add the execution listener to the intermediate catch event in your bpmn. This will let you perform a query on message subscription AND desired variables.