Do Zeebe clients only actively pull tasks after they have completed or failed? If there is only one task to be pulled at a time, the jobhandler executing this task will encounter blockages or take a long time. The zeebe client will not actively pull new tasks until the task fails or is completed. Doesn’t this lead to thread redundancy? Can other newly created tasks not be processed in a timely manner?
zeebe version 1.0.0
I’m not sure I understand your question.
Lets assume for a moment that you have 1 process definition, with 10 sequential activities in it (A-J). Let’s also assume for a moment that you have 10 instances (0-9) of this process.
Activity B0 cannot be fetched until after activity A0 has been completed.
Activity I7 cannot be fetched until after activity H7 has been completed.
If Activities E2, E4, and E8 are all ready at the same time, then one worker may collect all of them at the same time. If there are more “E” activities ready at the same time than the worker can work on at one time, then a second “E-Worker” could collect others.
If the E-Worker is busy working on all of E2, E4, and E8 at the same time, there is nothing stopping the J-Worker from working on J1, J5, and J9 … If those process instances are at that step.
You are on an EXTREMEMLY old version of Zeebe. You may be running into code limitations that have since been resolved. You should work towards upgrading.
If you can explain your questions more clearly, include BPMN diagrams if needed, the community will be happy to help clarify, but it’s not quite clear what specific instance-case you’re talking about.
A follow up, related to your other thread … If a worker (say the one working on E2, E4, E8) goes wayward, it won’t release the resources that it has consumed until it gets to its release point, even if the engine has decided that the worker is “missing in action” and allowed E-Worker #2 to collect E2, E4, and E8.
E-Worker#1 will never collect any more tasks.
If E-Worker#2 isn’t running, no E tasks will be completed.
package com.demo.adapt;
import java.time.Duration;
import java.util.concurrent.TimeUnit;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import io.camunda.zeebe.client.ZeebeClient;
import io.camunda.zeebe.client.api.response.ActivatedJob;
import io.camunda.zeebe.client.api.worker.JobClient;
import io.camunda.zeebe.client.api.worker.JobHandler;
import io.camunda.zeebe.client.api.worker.JobWorker;
import lombok.extern.slf4j.Slf4j;
@Slf4j
@Component
public class TestWorker{
@Autowired private ZeebeClient client;
private JobWorker worker;
@PostConstruct
public void register() {
worker = client
.newWorker()
.jobType("test")
.handler(new JobHandler() {
@Override
public void handle(JobClient client, ActivatedJob job) throws Exception {
log.info(job.getProcessInstanceKey() + "_" + job.getKey() +"_"+ System.currentTimeMillis());
TimeUnit.MINUTES.sleep(10);
}
})
.name("test")
.open();
log.info("Job worker test opened and receiving jobs");
}
// @Override
// public void handle(JobClient client, ActivatedJob job) throws Exception {
// System.err.println("currentTime:" + System.currentTimeMillis());
// TimeUnit.MINUTES.sleep(6);
// client.newCompleteCommand(job.getKey()).send();
// }
@PreDestroy
public void unregister() {
if (!worker.isClosed()) {
worker.close();
log.info("test Job worker closed");
}
}
}
<?xml version="1.0" encoding="UTF-8"?>
<bpmn:definitions xmlns:bpmn="http://www.omg.org/spec/BPMN/20100524/MODEL" xmlns:bpmndi="http://www.omg.org/spec/BPMN/20100524/DI" xmlns:dc="http://www.omg.org/spec/DD/20100524/DC" xmlns:zeebe="http://camunda.org/schema/zeebe/1.0" xmlns:di="http://www.omg.org/spec/DD/20100524/DI" id="Definitions_199dbsq" targetNamespace="http://bpmn.io/schema/bpmn" exporter="Zeebe Modeler" exporterVersion="0.11.0">
<bpmn:process id="test0003" name="test0003" isExecutable="true">
<bpmn:startEvent id="StartEvent_1">
<bpmn:outgoing>Flow_1dc6wpf</bpmn:outgoing>
</bpmn:startEvent>
<bpmn:sequenceFlow id="Flow_1dc6wpf" sourceRef="StartEvent_1" targetRef="Activity_0wjwjvs" />
<bpmn:serviceTask id="Activity_0wjwjvs" name="test">
<bpmn:extensionElements>
<zeebe:taskDefinition type="test" />
</bpmn:extensionElements>
<bpmn:incoming>Flow_1dc6wpf</bpmn:incoming>
<bpmn:outgoing>Flow_0n5l2r9</bpmn:outgoing>
</bpmn:serviceTask>
<bpmn:endEvent id="Event_0cwjxrj">
<bpmn:incoming>Flow_0n5l2r9</bpmn:incoming>
</bpmn:endEvent>
<bpmn:sequenceFlow id="Flow_0n5l2r9" sourceRef="Activity_0wjwjvs" targetRef="Event_0cwjxrj" />
</bpmn:process>
<bpmndi:BPMNDiagram id="BPMNDiagram_1">
<bpmndi:BPMNPlane id="BPMNPlane_1" bpmnElement="test0003">
<bpmndi:BPMNEdge id="Flow_1dc6wpf_di" bpmnElement="Flow_1dc6wpf">
<di:waypoint x="195" y="120" />
<di:waypoint x="350" y="120" />
</bpmndi:BPMNEdge>
<bpmndi:BPMNEdge id="Flow_0n5l2r9_di" bpmnElement="Flow_0n5l2r9">
<di:waypoint x="450" y="120" />
<di:waypoint x="632" y="120" />
</bpmndi:BPMNEdge>
<bpmndi:BPMNShape id="_BPMNShape_StartEvent_2" bpmnElement="StartEvent_1">
<dc:Bounds x="159" y="102" width="36" height="36" />
</bpmndi:BPMNShape>
<bpmndi:BPMNShape id="Activity_1y29qzv_di" bpmnElement="Activity_0wjwjvs">
<dc:Bounds x="350" y="80" width="100" height="80" />
</bpmndi:BPMNShape>
<bpmndi:BPMNShape id="Event_0cwjxrj_di" bpmnElement="Event_0cwjxrj">
<dc:Bounds x="632" y="102" width="36" height="36" />
</bpmndi:BPMNShape>
</bpmndi:BPMNPlane>
</bpmndi:BPMNDiagram>
</bpmn:definitions>
@Bean
public ZeebeClient zeebeClient() {
ZeebeClientBuilder clientBuilder = ZeebeClient.newClientBuilder();
clientBuilder.gatewayAddress("localhost:26500");
clientBuilder.numJobWorkerExecutionThreads(100);
clientBuilder.usePlaintext();
ZeebeClient client = clientBuilder.build();
return client;
}
@RequestMapping("/submit")
public String submit() {
Map<String, Object> map = new HashMap<String, Object>();
map.put("submitTime", System.currentTimeMillis());
ProcessInstanceEvent join = client.newCreateInstanceCommand().bpmnProcessId("test0003").latestVersion()
.variables(map).send().join();
log.info("Workflow instance created. Key: " + join.getProcessInstanceKey());
return "ok";
}
2023-07-06 10:22:46.897 INFO 4168 --- [nio-8080-exec-2] com.demo.controller.TestController : Workflow instance created. Key: 2251799831883027
2023-07-06 10:22:47.194 INFO 4168 --- [pool-2-thread-2] com.demo.adapt.TestWorker : 2251799831883027_2251799831883033_1688610167194
2023-07-06 10:22:48.005 INFO 4168 --- [nio-8080-exec-4] com.demo.controller.TestController : Workflow instance created. Key: 2251799831883035
My test code is as follows: submit two instances, but the sleep of the first job affects the job of the second instance. As expected, the job of the second instance should first print the log before proceeding with the sleep. But the job of the second instance did not output logs. If configuration is necessary to ensure that the jobs between instances do not affect each other when there are sufficient threads.
You have to be VERY careful with language here.
What I’m seeing (and your post is still VERY confusing) is that you have submitted two process instances, but have only one worker. Since it’s the worker that is writing the log entry, it wrote its log entry and went to sleep:
Workers are not tied to process instances… same way that you are not tied to just one project.