4. Writing Job Workers

Service tasks are done by job workers: your code, in any language, that asks the engine for work. This note covers how that works, how to handle success, failure and business errors, and how to write workers that are safe to run at scale. Architecture: Camunda 8 Architecture. Practice: Running Camunda 8 Locally.

The job lifecycle

stateDiagram-v2
    [*] --> Created: process reaches a service task
    Created --> Activated: worker activates it (lock for a timeout)
    Activated --> Completed: worker completes with variables
    Activated --> Failed: worker fails (retries minus 1)
    Failed --> Created: retries left, after back-off
    Failed --> Incident: no retries left
    Activated --> Created: timeout expired, job is offered again
    Activated --> ErrorThrown: worker throws a BPMN error
    Incident --> Created: you fix the cause and retry
    Completed --> [*]
    ErrorThrown --> [*]: caught by a boundary event

• The job type string in the model (check-inventory) is how a worker is matched to a task.
• A worker activates jobs with a timeout: if it does not finish in time, the job is offered to another worker.
• Workers use long polling: they hold a request open until a job is available, so there is little delay and little wasted traffic.

Three ways to finish a job

Rule of thumb: fail for things that might work if tried again; throw an error for outcomes the business has a path for.

A worker, in three languages

Node.js (TypeScript)

import { Camunda8 } from '@camunda8/sdk'
const zeebe = new Camunda8().getZeebeGrpcApiClient()

zeebe.createWorker({
  taskType: 'check-inventory',
  taskHandler: async (job) => {
    const { sku, quantity } = job.variables
    const stock = await inventory.lookup(sku)          // your own code
    if (stock < quantity) {
      return job.error('out_of_stock', `Only ${stock} left`)   // business error
    }
    return job.complete({ reserved: true })
  },
})

Python (pyzeebe)

@worker.task(task_type="charge-payment", timeout_ms=30_000)
async def charge(order_id: str, amount: float) -> dict:
    result = await payments.charge(order_id, amount)   # raises on technical errors
    if result.declined:
        raise BusinessError("payment_failed")          # mapped to a BPMN error
    return {"paymentId": result.id}

Java (Spring Boot with the Camunda starter)

@JobWorker(type = "ship-items")
public Map<String, Object> ship(@Variable String orderId) {
    String trackingNumber = shipping.createShipment(orderId);
    return Map.of("trackingNumber", trackingNumber);
}

Writing good workers

Tuning knobs

Messages and correlation

How the outside world talks to a running instance:

sequenceDiagram
    participant X as Your system
    participant Z as Zeebe
    X->>Z: Publish message "payment_confirmed" with correlationKey "order-1001"
    Z->>Z: Find the instance waiting for this message with correlationKey = order-1001
    Z-->>X: Correlated, instance continues

• A message start event creates a new instance; intermediate catch events and message boundary events correlate to existing ones.
• Each waiting event has a correlation key expression (for example =order_id); the message's key must equal that value.
• Message time-to-live (TTL) lets a message wait a little for an instance that has not reached the event yet. A message ID prevents duplicates being published twice.
• Signals broadcast to many instances at once.

Connectors: no code for common integrations

For REST calls, Slack, email, Kafka, SQL and so on, add a connector to a task in the Modeler and fill in its fields, with secrets referenced as {{secrets.MY_KEY}}. Write a worker only when you need your own logic. Where possible, prefer a connector for plumbing and a worker for business logic.

Starting instances and publishing messages

$ curl -u demo:demo -X POST http://localhost:8080/v2/process-instances \
    -H "Content-Type: application/json" \
    -d '{"processDefinitionId":"order-process","variables":{"orderId":"A-1001"}}'

$ curl -u demo:demo -X POST http://localhost:8080/v2/messages/publication \
    -H "Content-Type: application/json" \
    -d '{"name":"payment_confirmed","correlationKey":"A-1001","variables":{"paid":true}}'

Security

• Use a dedicated API client per worker, with only the permissions it needs.
• Keep client secrets in environment variables or a secret store, never in the repo.
• Workers call out to the engine, so they need no inbound ports. Run them in a private network.
• Do not log variables that hold personal data. See Secure Coding and Supply Chain.