Hi everyone.
I’m using Camunda to start process instances based on asynchronous events / notifications of an external system. Multiple notifications can relate to the same topic, but contain different bodies. The notifications may or may not be able to be processed automatically (Process notification is asyncBefore):
After processing a notification, a new process should be started that processes its topic and eventually does something with it, let’s say writing an E-Mail about it. Because potentially multiple notifications may reference this topic, the process must be able to aggregate new results. After each new result, some heuristics are performed to determine whether or not to wait for more results. A bare-bones version of this might be the following:
Obviously this will only work if there’s only one instance per topic of the second process. The current delegate code of my message end event in the first process looks like this:
@Override
public void execute(final DelegateExecution execution) throws Exception {
// Look for active process instances for the same topic
final ProcessInstance processInstance = runtimeService.createProcessInstanceQuery()
.processDefinitionKey("Process_1mkl685")
.variableValueEquals("topicName", "myTopicName")
.active()
.singleResult(); // There mustn't be more than one active process for this topic
if (processInstance == null) {
runtimeService.startProcessInstanceByMessage("Message_2ou8g3l", Map.of(
"topicName", "myTopicName"
// Result of notification processing...
));
} else {
// Deliver message to running instance...
}
}
Everything is fine as long as most notifications must be processed manually (as in, there’s a significant delay between process creation and completion). However, once multiple notifications about the same topic arrive that all can be processed automatically, the ProcessInstanceQuery fails with:
org.camunda.bpm.engine.ProcessEngineException: Query return 3 results instead of max 1
Coincidentally, 3 is also the core size of the job execution pool. So all 3 jobs ran in parallel, none of them finding any instances for the given topic, leading to them starting new ones. When the next 3 jobs arrive at this point, they fail with the error above. Is there a sane solution for this?
An alternative approach I was playing around with was to offload the aggregation to an external task worker that would start the second process via REST API. This way, Camunda acts as a queue for results of the first process. However, I’m curious if this can be solved with Camunda alone?