Camunda 8 Service Task Contract & Retry Backoff Gate

id: camunda-job-worker-contract-and-retry-backoff-gate
namespace: company.team

description: |
  Act as a resilient external job worker for Camunda 8 (Zeebe) BPMN service tasks.
  Validates incoming job payloads against required schema contracts, completes valid tasks
  with enriched output variables, applies exponential retry backoff on transient failures,
  and raises immediate incidents in Camunda Operate when retries are exhausted.

inputs:
  - id: camunda_rest_address
    type: STRING
    displayName: Camunda REST Address
    description: Base URL of the Camunda 8 Zeebe gateway REST API (e.g.,
      http://localhost:8080 for local Zeebe or SaaS endpoint).
    defaults: "http://localhost:8080"

  - id: camunda_grpc_address
    type: STRING
    displayName: Camunda gRPC Address
    description: Host and port of the Camunda Zeebe gRPC gateway (e.g., localhost:26500).
    defaults: "localhost:26500"

  - id: job_type
    type: STRING
    displayName: BPMN Service Task Job Type
    description: The task definition type configured on the Camunda BPMN Service
      Task element.
    defaults: "process-order"

  - id: worker_name
    type: STRING
    displayName: Worker Name
    description: Client identifier registered with Zeebe when activating job locks.
    defaults: "kestra-camunda-worker"

  - id: lock_timeout
    type: DURATION
    displayName: Job Lock Timeout
    description: Maximum duration the job lock is held by Kestra before Zeebe makes
      it available to other workers.
    defaults: "PT5M"

  - id: retry_backoff
    type: DURATION
    displayName: Retry Backoff Duration
    description: ISO-8601 duration Camunda waits before re-activating a transiently
      failed job.
    defaults: "PT30S"

triggers:
  - id: camunda_job_stream
    type: io.kestra.plugin.camunda.Trigger
    description: "External job worker listening for activated BPMN service tasks of
      type jobType via REST long polling. Shipped disabled so connection
      parameters and secrets can be configured first."
    restAddress: "{{ inputs.camunda_rest_address }}"
    grpcAddress: "{{ inputs.camunda_grpc_address }}"
    jobType: "{{ inputs.job_type }}"
    workerName: "{{ inputs.worker_name }}"
    timeout: PT5M
    streamEnabled: false
    disabled: true

tasks:
  - id: validate_payload_contract
    type: io.kestra.plugin.core.debug.Return
    description: "Validate that the incoming Camunda job variables contain the
      mandatory business fields (order_id, customer_id, and positive amount)."
    format: "{{ trigger.variables.order_id is defined and
      trigger.variables.customer_id is defined and trigger.variables.amount is
      defined and trigger.variables.amount > 0 ? 'VALID_CONTRACT' :
      'INVALID_SCHEMA' }}"

  - id: route_by_contract_evaluation
    type: io.kestra.plugin.core.flow.If
    description: "Branch on payload validity: execute business fulfillment and
      complete the job, or immediately fail with zero retries on invalid
      contract."
    condition: "{{ outputs.validate_payload_contract.value == 'VALID_CONTRACT' }}"
    then:
      - id: log_job_processing
        type: io.kestra.plugin.core.log.Log
        description: "Log validated job processing with process instance and element keys."
        message: "Processing Camunda jobKey={{ trigger.jobKey }} for processInstance={{
          trigger.processInstanceKey }}, bpmnProcess={{ trigger.bpmnProcessId
          }}, elementId={{ trigger.elementId }}."

      - id: complete_service_task
        type: io.kestra.plugin.camunda.CompleteJob
        description: "Report the job as successfully completed back to Camunda Zeebe,
          merging enriched variables into the process instance."
        restAddress: "{{ inputs.camunda_rest_address }}"
        jobKey: "{{ trigger.jobKey }}"
        variables:
          fulfillment_status: "SUCCESS"
          processed_by: "kestra-worker"
          processed_at: "{{ now() }}"

      - id: log_job_completed
        type: io.kestra.plugin.core.log.Log
        message: "Successfully completed Camunda job {{ trigger.jobKey }} and released
          lock."

    else:
      - id: fail_job_permanent_invalid_schema
        type: io.kestra.plugin.camunda.FailJob
        description: "Explicitly fail the job with 0 retries so Camunda Operate raises
          an immediate actionable Incident rather than letting the lock expire
          silently into a zombie loop."
        restAddress: "{{ inputs.camunda_rest_address }}"
        jobKey: "{{ trigger.jobKey }}"
        retries: 0
        errorMessage: "Schema validation contract violated: missing or invalid order_id,
          customer_id, or non-positive amount. Variables received: {{
          trigger.variables }}."
        variables:
          failure_reason: "CONTRACT_VIOLATION"

      - id: alert_invalid_contract_incident
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: "Notify on-call team in Slack that an unrecoverable contract
          violation occurred and an incident was created in Operate."
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "CRITICAL: Camunda Job {{ trigger.jobKey }} in process {{ trigger.bpmnProcessId }} (instance {{ trigger.processInstanceKey }}) failed schema validation contract. Job failed with 0 retries; incident created in Camunda Operate. Execution: {{ execution.id }}."
          }

  - id: log_worker_audit
    type: io.kestra.plugin.core.log.Log
    description: "Record immutable audit line for operational visibility."
    message: "Camunda Worker Audit: jobKey={{ trigger.jobKey }}, bpmnProcess={{
      trigger.bpmnProcessId }}, contractStatus={{
      outputs.validate_payload_contract.value }}, executionId={{ execution.id
      }}."

errors:
  - id: fail_job_transient_error
    type: io.kestra.plugin.camunda.FailJob
    description: "Safely decrement job retries and set retry backoff in Camunda when
      an unexpected worker error occurs. If retries hit 0, Camunda raises an
      incident."
    restAddress: "{{ inputs.camunda_rest_address }}"
    jobKey: "{{ trigger.jobKey }}"
    retries: "{{ trigger.retries is defined and trigger.retries > 1 ?
      trigger.retries - 1 : 0 }}"
    retryBackoff: "{{ inputs.retry_backoff }}"
    errorMessage: "Kestra execution {{ execution.id }} encountered an unexpected
      worker error while executing job {{ trigger.jobKey }}."

  - id: alert_worker_execution_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: "Alert Slack when an unexpected worker failure occurs."
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "ALERT: Kestra Camunda worker execution {{ execution.id }} failed on job {{ trigger.jobKey }}. Remaining retries: {{ trigger.retries is defined and trigger.retries > 1 ? trigger.retries - 1 : 0 }}."
      }

outputs:
  - id: job_key
    type: STRING
    value: "{{ trigger.jobKey }}"

  - id: process_instance_key
    type: STRING
    value: "{{ trigger.processInstanceKey }}"

  - id: contract_evaluation
    type: STRING
    value: "{{ outputs.validate_payload_contract.value }}"