Temporal Workers polling from a queue

Community Support
1

Im designing a system using the java sdk where I have different temporal workers in different locations listening to a certain task pool. When a workflow is started, as the first workflow activity each worker polls a message queue which responds with data based on the worker’s location and then the worker processes it. The second activity is receiving the processed data and sends it to DB. What is the best way to make the workers continually poll the message queue? I considered having the workflow activities in a while loop, but in that case when a certain worker exits the loop, temporal considers the workflow complete and stops all workers.

2

Looks like you need to route all activities to same worker, for that can take a look at fileprocessing sample.

but in that case when a certain worker exits the loop, temporal considers the workflow complete and stops all workers.

note sure i fully understand the “and stops all workers” part, can you give more info please?

3

To test I have 2 instances of worker started. When I call WorkflowClient.start(), both workers start consuming from the queue, as expected. I encounter situations when the queue has no more messages for one of the workers, but still has messages for the other. The worker for which there are no more messages exits the while loop and the workflow is marked complete, therefore the second worker also stops.

4

Could you share your code where you create workers and start executions (and this loop you mention).

When I call WorkflowClient.start(), both workers start consuming from the queue, as expected

workers start polling as soon as you start the WorkerFactory. you can stop your workers by calling shutdown method on the worker factory.

The worker for which there are no more messages exits the while loop

would like to understand this better please, not sure why while loop is needed and what it does. thanks.

5

Cant share code. Can share pseudocode :slight_smile:

This is the workflow implementation that is ran by both workers.

Activity One {
 HTTP Post Request to Server to get worker specific data
 Process Data
}

Activity Two {
 Send Processed Data to DB
}

Workflow {
 while(true) {
 ActivityOne
 if(processedData.notEmpty) {
 ActivityTwo}
else{
break
}