Camunda8 parallel multi instance concurrency issue with springboot job worker. I have a parent process and a sub-process into it.
The sub-process is a parralel multi intsance task. The sub-process has a child process into it.
I have an array declared in parent process service task and it is being updated in the sub-process. But the array is not getting updated as it should.
The initiazed array is coming to the update again and again (if an element is already updated, it is not visible in the array). So, in the end only the last element of the array is updated in final result.
I need help to resolve the issue. Please help.
Here are my bpmn processes and the job worker code.
camunda8_processElement.bpmn (4.6 KB)
camunda8_workfloC.bpmn (13.7 KB)
import io.camunda.zeebe.client.api.response.ActivatedJob;
import io.camunda.zeebe.client.api.worker.JobClient;
import io.camunda.zeebe.spring.client.annotation.JobWorker;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;
@Component
public class ZeebeComponents {
private static final Logger log = LoggerFactory.getLogger(ZeebeComponents.class);
private static final String scriptTask_initialize = "scriptTask-initialize";
private static final String scriptTask_selectFirstElement = "scriptTask-selectFirstElement";
private static final String subScriptTask_initializeElement = "scriptSubTask-initializeElement";
private static final String subScriptTask_updateElement = "scriptSubTask-updateElement";
private static final String subScriptTask_updateResult = "scriptSubTask-updateResult";
private static final String subScriptTask_selectNextElement = "scriptSubTask-selectNextElement";
private static final String scriptTask_logFinalResult = "scriptTask-logFinalResult";
private static final String scriptTask_reportCurrentResult = "scriptTask-reportCurrentResult";
private static final String scriptTask_reportError = "scriptTask-reportError";
private static final String scriptTask_reportTimeoutError = "scriptTask-reportTimeoutError";
private List<String> results = null;
@JobWorker(type = scriptTask_initialize)
public Map<String, Object> handleScriptTask01(final ActivatedJob job) {
Map<String, Object> variables = job.getVariablesAsMap();
Map<String, Object> configuration = (Map<String, Object>) variables.get("configuration");
Integer numElements = (Integer) configuration.get("numElements");
Integer elementSleepSeconds = (Integer) configuration.get("elementSleepSeconds");
variables.put("numElements", numElements);
variables.put("elementSleepSeconds", elementSleepSeconds);
variables.put("startTime", LocalDateTime.now().toString());
String [] elementsArr = new String[numElements];
results = new CopyOnWriteArrayList<>();
for (int i = 0; i < numElements; i++){
results.add("??");
elementsArr[i] = "element"+i;
}
variables.put("elements", elementsArr);
variables.put("results", results);
return variables;
}
@JobWorker(type = subScriptTask_initializeElement)
public Map<String, Object> handleScriptTask02(ActivatedJob job) {
Map<String, Object> variables = job.getVariablesAsMap();
String sleepDuration = "PT0S";
Integer elementSleepSeconds = (Integer) variables.get("elementSleepSeconds");
String element = (String) variables.get("element");
log.info("Processing element -> "+element);
if(elementSleepSeconds > 0){
sleepDuration = "PT"+elementSleepSeconds+"S";
}
variables.put("sleepDuration", sleepDuration);
return variables;
}
@JobWorker(type = subScriptTask_updateElement)
public Map<String, Object> handleScriptTask03( ActivatedJob job) {
Map<String, Object> variables = job.getVariablesAsMap();
variables.put("elementResult", "success");
return variables;
}
@JobWorker(type = subScriptTask_updateResult)
synchronized public Map<String, Object> handleScriptTask05(JobClient client, ActivatedJob job) {
Map<String, Object> variables = job.getVariablesAsMap();
String element = (String) variables.get("element");
int index = Integer.parseInt(element.substring(7));
String elementResult = (String) variables.get("elementResult");
List<String> results = (List<String>) variables.get("results");
synchronized (results){
results.set(index, elementResult);
}
variables.put("results", results);
log.info(subScriptTask_updateResult+" [Results] -> "+results.toString());
return variables;
}
@JobWorker(type = scriptTask_logFinalResult)
public void reportFinalResult (final ActivatedJob job) {
Map<String, Object> variables = job.getVariablesAsMap();
log.info(variables.toString());
log.info("=================================================================");
String startTime = (String) variables.get("startTime");
LocalDateTime endTime = LocalDateTime.now();
Duration duration = Duration.between(endTime, LocalDateTime.parse(startTime));
log.info("Execution Time in milli seconds: "+duration.toMillis());
log.info("=================================================================");
}
@JobWorker(type = scriptTask_reportCurrentResult)
public void logCurrentResult (final ActivatedJob job) {
log.info("---------------------------------------------Current Result--------------------------------------------------------------------");
Map<String, Object> variables = job.getVariablesAsMap();
List<String> strArrList = (List<String>) variables.get("results");
log.info(strArrList.toString());
log.info("-------------------------------------------------------------------------------------------------------------------------------");
}
@JobWorker(type = scriptTask_reportError)
public void logError () {
log.info("Error!!!");
}
@JobWorker(type = scriptTask_reportTimeoutError)
public void logTimeOutError () {
log.info("Error due to Timeout!!!");
}
}