Camunda Missing Message Correlation Backfill with Kestra

id: camunda-missing-message-correlation-backfill
namespace: company.team
description: |
  Find active Camunda process instances that have waited past an SLA on an allowed message catch
  while a matching unsent event remains in a PostgreSQL outbox. In report-only mode the flow shows
  the instance keys, message names and correlation keys. With both safety inputs explicitly enabled,
  it re-reads the plan after approval, republishes each event once through PublishMessage with a
  serial loop, then proves that no owned instance still has the matching message subscription.

concurrency:
  limit: 1
  behavior: QUEUE

inputs:
  - id: camunda_rest_address
    type: STRING
    defaults: http://localhost:8080
    description: Camunda 8 REST API base URL for the recovery cluster.
  - id: camunda_username
    type: STRING
    defaults: ''
    description: Optional basic-auth username.
  - id: camunda_password
    type: STRING
    defaults: ''
    description: Optional basic-auth password.
  - id: outbox_jdbc_url
    type: STRING
    defaults: jdbc:postgresql://localhost:5432/outbox
    description: PostgreSQL JDBC URL for the producer outbox.
  - id: outbox_username
    type: STRING
    defaults: outbox
    description: PostgreSQL outbox reader username.
  - id: outbox_password
    type: STRING
    defaults: outbox
    description: PostgreSQL outbox reader password.
  - id: process_definition_id
    type: STRING
    defaults: order_fulfillment
    description: One exact BPMN process ID allowed to be repaired.
  - id: message_name
    type: STRING
    defaults: order-payment-authorized
    description: One exact message name allowed to be republished.
  - id: sla_minutes
    type: INT
    defaults: 15
    min: 1
    max: 10080
    description: Candidate instances and outbox rows must both be older than this window.
  - id: max_candidates
    type: INT
    defaults: 10
    min: 1
    max: 100
    description: Hard ceiling for one run. A larger finding set refuses all writes.
  - id: auto_remediate
    type: BOOL
    defaults: false
    description: Permit republishing only when true.
  - id: dry_run
    type: BOOL
    defaults: true
    description: Keep the run read-only unless explicitly false together with
      auto_remediate.

variables:
  candidate_query: |
    {{ {'subscriptions': outputs.read_subscriptions.body, 'instances': outputs.read_instances.body, 'outbox': outputs.read_outbox.rows, 'max_candidates': inputs.max_candidates, 'auto_remediate': inputs.auto_remediate, 'dry_run': inputs.dry_run} | toJson
        | jq('. as $r | (($r.subscriptions | fromjson).items // []) as $s | (($r.instances | fromjson).items // []) as $i | ($r.outbox // []) as $o | [$s[] as $sub | ($i[] | select((.processInstanceKey | tostring) == ($sub.processInstanceKey | tostring))) as $inst | ($o[] | select((.message_name | tostring) == ($sub.messageName | tostring) and (.correlation_key | tostring) == ($sub.correlationKey | tostring))) as $event | {instance_key: ($sub.processInstanceKey | tostring), definition_id: ($inst.processDefinitionId | tostring), message_name: ($sub.messageName | tostring), correlation_key: ($sub.correlationKey | tostring), event_id: ($event.event_id | tostring), payload: (($event.payload // "{}") | tostring), instance_created: ($inst.startDate // $inst.creationTime // ""), event_created: ($event.created_at | tostring)}] as $c | {candidates: $c, count: ($c | length), signature: ([$c[] | (.instance_key + "|" + .message_name + "|" + .correlation_key + "|" + .event_id)] | sort | join(",")), decision: (if ($c | length) == 0 then "NO_FINDINGS" elif ($c | length) > $r.max_candidates then "OVER_CEILING" elif ($r.auto_remediate and ($r.dry_run | not)) then "REMEDIATE" else "REPORT_ONLY" end)} | tojson') | first }}

tasks:
  - id: read_subscriptions
    type: io.kestra.plugin.core.http.Request
    description: Read CREATED message subscriptions for the one allowed process and
      message name.
    uri: "{{ inputs.camunda_rest_address }}/v2/message-subscriptions/search"
    method: POST
    contentType: application/json
    body: |
      {
        "filter": {
          "messageName": "{{ inputs.message_name }}",
          "processDefinitionId": "{{ inputs.process_definition_id }}",
          "messageSubscriptionState": "CREATED"
        },
        "page": {"limit": 1000}
      }

  - id: read_instances
    type: io.kestra.plugin.core.http.Request
    description: Read only ACTIVE instances of the exact allowed process that
      started before the SLA cutoff.
    uri: "{{ inputs.camunda_rest_address }}/v2/process-instances/search"
    method: POST
    contentType: application/json
    body: |
      {
        "filter": {
          "state": "ACTIVE",
          "processDefinitionId": "{{ inputs.process_definition_id }}",
          "startDate": {"$lt": "{{ execution.startDate | dateAdd(0 - inputs.sla_minutes, 'MINUTES') | date("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'", timeZone="UTC") }}"}
        },
        "page": {"limit": 1000}
      }

  - id: read_outbox
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: Read old pending producer events. The query intentionally does not
      mutate the outbox.
    url: "{{ inputs.outbox_jdbc_url }}"
    username: "{{ inputs.outbox_username }}"
    password: "{{ inputs.outbox_password }}"
    fetchType: FETCH
    sql: |
      SELECT event_id::text AS event_id,
             message_name,
             correlation_key,
             payload::text AS payload,
             created_at::text AS created_at
        FROM outbox_events
       WHERE status = 'PENDING'
         AND published_at IS NULL
         AND message_name = '{{ inputs.message_name }}'
         AND created_at < now() - interval '{{ inputs.sla_minutes }} minutes'
       ORDER BY created_at
       LIMIT {{ inputs.max_candidates + 1 }}

  - id: classify
    type: io.kestra.plugin.core.output.OutputValues
    description: Intersect the engine's parked subscriptions with old active
      instances and old pending outbox rows.
    values:
      result: >-
        {{ {'subscriptions': outputs.read_subscriptions.body, 'instances':
        outputs.read_instances.body, 'outbox': outputs.read_outbox.rows,
        'max_candidates': inputs.max_candidates, 'auto_remediate':
        inputs.auto_remediate, 'dry_run': inputs.dry_run} | toJson
          | jq('. as $r | (($r.subscriptions | fromjson).items // []) as $s | (($r.instances | fromjson).items // []) as $i | ($r.outbox // []) as $o | [$s[] as $sub | ($i[] | select((.processInstanceKey | tostring) == ($sub.processInstanceKey | tostring))) as $inst | ($o[] | select((.message_name | tostring) == ($sub.messageName | tostring) and (.correlation_key | tostring) == ($sub.correlationKey | tostring))) as $event | {instance_key: ($sub.processInstanceKey | tostring), definition_id: ($inst.processDefinitionId | tostring), message_name: ($sub.messageName | tostring), correlation_key: ($sub.correlationKey | tostring), event_id: ($event.event_id | tostring), payload: (($event.payload // "{}") | tostring), instance_created: ($inst.startDate // $inst.creationTime // ""), event_created: ($event.created_at | tostring)}] as $c | {candidates: $c, count: ($c | length), signature: ([$c[] | (.instance_key + "|" + .message_name + "|" + .correlation_key + "|" + .event_id)] | sort | join(",")), decision: (if ($c | length) == 0 then "NO_FINDINGS" elif ($c | length) > $r.max_candidates then "OVER_CEILING" elif ($r.auto_remediate and ($r.dry_run | not)) then "REMEDIATE" else "REPORT_ONLY" end)} | tojson') | first }}

  - id: log_plan
    type: io.kestra.plugin.core.log.Log
    message: >-
      Camunda missing-message scan: subscriptions={{
      outputs.read_subscriptions.body | jq('.page.totalItems') | first }}
      active_past_sla={{ outputs.read_instances.body | jq('.page.totalItems') |
      first }} pending_outbox={{ outputs.read_outbox.size }} candidates={{
      outputs.classify.values.result | jq('.count') | first }} decision={{
      outputs.classify.values.result | jq('.decision') | first }} keys={{
      outputs.classify.values.result | jq('[.candidates[] | (.instance_key + "="
      + .correlation_key)] | join(", ")') | first }}

  - id: route
    type: io.kestra.plugin.core.flow.Switch
    value: "{{ outputs.classify.values.result | jq('.decision') | first }}"
    cases:
      NO_FINDINGS:
        - id: log_no_findings
          type: io.kestra.plugin.core.log.Log
          message: No eligible parked instances have a matching old pending outbox event.
            No Camunda or PostgreSQL write was attempted.
      OVER_CEILING:
        - id: fail_over_ceiling
          type: io.kestra.plugin.core.execution.Fail
          errorMessage: "{{ outputs.classify.values.result | jq('.count') | first }}
            candidates exceed max_candidates={{ inputs.max_candidates }}.
            Nothing was republished. Narrow the allowlist or investigate the
            producer outage."
      REPORT_ONLY:
        - id: log_dry_run
          type: io.kestra.plugin.core.log.Log
          message: "Read-only plan: {{ outputs.classify.values.result | jq('.count') |
            first }} message(s) would be republished. auto_remediate={{
            inputs.auto_remediate }} dry_run={{ inputs.dry_run }}."
      REMEDIATE:
        - id: approval
          type: io.kestra.plugin.core.flow.Pause
          pauseDuration: PT30M
          behavior: FAIL
          description: Republish only the exact instance, message and outbox keys shown in
            log_plan. This sends external business events and cannot be rolled
            back.
          onResume:
            - id: approved
              type: BOOL
              defaults: false
              description: Answer true only after reviewing the bounded candidate list.
        - id: approval_gate
          type: io.kestra.plugin.core.flow.If
          condition: "{{ outputs.approval.onResume.approved ?? false }}"
          then:
            - id: reread_subscriptions
              type: io.kestra.plugin.core.http.Request
              description: Re-read subscriptions immediately before each write phase so
                approval never authorizes a stale plan.
              uri: "{{ inputs.camunda_rest_address }}/v2/message-subscriptions/search"
              method: POST
              contentType: application/json
              body: |
                {"filter":{"messageName":"{{ inputs.message_name }}","processDefinitionId":"{{ inputs.process_definition_id }}","messageSubscriptionState":"CREATED"},"page":{"limit":1000}}
            - id: reread_instances
              type: io.kestra.plugin.core.http.Request
              uri: "{{ inputs.camunda_rest_address }}/v2/process-instances/search"
              method: POST
              contentType: application/json
              body: |
                {"filter":{"state":"ACTIVE","processDefinitionId":"{{ inputs.process_definition_id }}","startDate":{"$lt":"{{ execution.startDate | dateAdd(0 - inputs.sla_minutes, 'MINUTES') | date("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'", timeZone="UTC") }}"}},"page":{"limit":1000}}
            - id: reread_outbox
              type: io.kestra.plugin.jdbc.postgresql.Query
              url: "{{ inputs.outbox_jdbc_url }}"
              username: "{{ inputs.outbox_username }}"
              password: "{{ inputs.outbox_password }}"
              fetchType: FETCH
              sql: |
                SELECT event_id::text AS event_id, message_name, correlation_key, payload::text AS payload, created_at::text AS created_at
                  FROM outbox_events
                 WHERE status = 'PENDING' AND published_at IS NULL AND message_name = '{{ inputs.message_name }}'
                   AND created_at < now() - interval '{{ inputs.sla_minutes }} minutes'
                 ORDER BY created_at LIMIT {{ inputs.max_candidates + 1 }}
            - id: recheck
              type: io.kestra.plugin.core.output.OutputValues
              values:
                result: >-
                  {{ {'subscriptions': outputs.reread_subscriptions.body,
                  'instances': outputs.reread_instances.body, 'outbox':
                  outputs.reread_outbox.rows, 'max_candidates':
                  inputs.max_candidates, 'auto_remediate': true, 'dry_run':
                  false} | toJson | jq('. as $r | (($r.subscriptions |
                  fromjson).items // []) as $s | (($r.instances |
                  fromjson).items // []) as $i | ($r.outbox // []) as $o | [$s[]
                  as $sub | ($i[] | select((.processInstanceKey | tostring) ==
                  ($sub.processInstanceKey | tostring))) as $inst | ($o[] |
                  select((.message_name | tostring) == ($sub.messageName |
                  tostring) and (.correlation_key | tostring) ==
                  ($sub.correlationKey | tostring))) as $event | {instance_key:
                  ($sub.processInstanceKey | tostring), definition_id:
                  ($inst.processDefinitionId | tostring), message_name:
                  ($sub.messageName | tostring), correlation_key:
                  ($sub.correlationKey | tostring), event_id: ($event.event_id |
                  tostring), payload: (($event.payload // "{}") | tostring),
                  instance_created: ($inst.startDate // $inst.creationTime //
                  ""), event_created: ($event.created_at | tostring)}] as $c |
                  {candidates: $c, count: ($c | length), signature: ([$c[] |
                  (.instance_key + "|" + .message_name + "|" + .correlation_key
                  + "|" + .event_id)] | sort | join(",")), decision: (if ($c |
                  length) == 0 then "NO_FINDINGS" elif ($c | length) >
                  $r.max_candidates then "OVER_CEILING" else "REMEDIATE" end)} |
                  tojson') | first }}
            - id: stale_plan_guard
              type: io.kestra.plugin.core.flow.If
              condition: "{{ outputs.recheck.values.result | jq('.signature') | first !=
                outputs.classify.values.result | jq('.signature') | first }}"
              then:
                - id: stale_plan
                  type: io.kestra.plugin.core.execution.Fail
                  errorMessage: "Approval became stale: the live instance, subscription or outbox
                    set changed. No message was republished."
              else:
                - id: republish
                  type: io.kestra.plugin.core.flow.Loop
                  concurrencyLimit: 1
                  values: "{{ outputs.recheck.values.result | jq('.candidates | tojson') | first
                    }}"
                  tasks:
                    - id: publish_one
                      type: io.kestra.plugin.camunda.PublishMessage
                      description: Republish one missing event through Camunda's typed message
                        command.
                      restAddress: "{{ inputs.camunda_rest_address }}"
                      username: "{{ inputs.camunda_username }}"
                      password: "{{ inputs.camunda_password }}"
                      timeout: PT30S
                      messageName: "{{ fromJson(item.value).message_name }}"
                      correlationKey: "{{ fromJson(item.value).correlation_key }}"
                      messageId: "{{ fromJson(item.value).event_id }}"
                      timeToLive: PT0S
                      variables: "{{ fromJson(fromJson(item.value).payload) }}"
                - id: settle
                  type: io.kestra.plugin.core.flow.Sleep
                  duration: PT3S
                - id: verify_subscriptions
                  type: io.kestra.plugin.core.http.Request
                  description: Re-read the owned subscriptions after the write. A green
                    PublishMessage task is not proof that the waiting instance
                    moved.
                  uri: "{{ inputs.camunda_rest_address }}/v2/message-subscriptions/search"
                  method: POST
                  contentType: application/json
                  body: |
                    {"filter":{"messageName":"{{ inputs.message_name }}","processDefinitionId":"{{ inputs.process_definition_id }}","messageSubscriptionState":"CREATED"},"page":{"limit":1000}}
                - id: verify_instances
                  type: io.kestra.plugin.core.http.Request
                  uri: "{{ inputs.camunda_rest_address }}/v2/process-instances/search"
                  method: POST
                  contentType: application/json
                  body: |
                    {"filter":{"state":"ACTIVE","processDefinitionId":"{{ inputs.process_definition_id }}"},"page":{"limit":1000}}
                - id: verify_result
                  type: io.kestra.plugin.core.flow.If
                  condition: "{{ {'subscriptions': outputs.verify_subscriptions.body,
                    'candidates': outputs.recheck.values.result} | toJson |
                    jq('((.subscriptions | fromjson).items // []) as $s |
                    ((.candidates | fromjson).candidates // []) as $c | [$s[] as
                    $sub | $c[] | select((.instance_key | tostring) ==
                    ($sub.processInstanceKey | tostring))] | length') | first !=
                    0 }}"
                  then:
                    - id: fail_still_parked
                      type: io.kestra.plugin.core.execution.Fail
                      errorMessage: "The Camunda re-read still found a matching CREATED subscription
                        after PublishMessage. The run failed so the remaining
                        parked instance is visible for investigation."
                  else:
                    - id: log_verified
                      type: io.kestra.plugin.core.log.Log
                      message: "Republished and verified {{ outputs.recheck.values.result |
                        jq('.count') | first }} message(s). Matching CREATED
                        subscriptions after the write: 0. Active instances in
                        the allowed process: {{ outputs.verify_instances.body |
                        jq('.page.totalItems') | first }}."
          else:
            - id: log_declined
              type: io.kestra.plugin.core.log.Log
              message: Approval declined. No Camunda message was republished and the outbox
                was unchanged.
    defaults:
      - id: fail_unknown_decision
        type: io.kestra.plugin.core.execution.Fail
        errorMessage: Detector returned no recognized decision. Treat the cluster as
          unaudited.

  - id: complete_trigger_job
    type: io.kestra.plugin.core.flow.If
    description: Complete the optional Camunda service-task job only after the
      recovery proof succeeds.
    condition: "{{ trigger.jobKey is defined }}"
    then:
      - id: complete_job
        type: io.kestra.plugin.camunda.CompleteJob
        restAddress: "{{ inputs.camunda_rest_address }}"
        username: "{{ inputs.camunda_username }}"
        password: "{{ inputs.camunda_password }}"
        jobKey: "{{ trigger.jobKey }}"

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {"text":"Camunda missing-message correlation backfill FAILED in execution {{ execution.id }}. No green task should be treated as proof of recovery; inspect the independent process and subscription reads."}

triggers:
  - id: camunda_repair_job
    type: io.kestra.plugin.camunda.Trigger
    description: Disabled job worker for an explicitly modeled repair service task.
      Manual runs remain available for incident response.
    disabled: true
    restAddress: "{{ inputs.camunda_rest_address }}"
    username: "{{ inputs.camunda_username }}"
    password: "{{ inputs.camunda_password }}"
    jobType: camunda-missing-message-correlation-backfill
    timeout: PT15M
    maxJobsActive: 1
  - id: hourly_safe_scan
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "17 * * * *"
    disabled: true
    inputs:
      auto_remediate: false
      dry_run: true

Safety

Republishing an event can advance an order, charge a customer or trigger another external side effect. The flow therefore limits the target to one exact process definition and message name, requires both an old active message subscription and an old pending outbox row with the same correlation key, refuses a candidate set above max_candidates and defaults to read-only. Approval is a stale snapshot: the engine and outbox are read again immediately before the write. Every message uses the outbox event ID as Camunda's deduplication ID. A serial loop limits pressure on the broker. After publishing, the flow reads Camunda's subscriptions again and fails if any matching instance is still parked.

Guard	What it stops
Exact process and message inputs	Cross-workflow or cross-event repair
SLA on both engine and outbox rows	Repairing a fresh event or a recently started instance
Matching correlation key and event ID	Sending an event to the wrong order or twice
Candidate ceiling and serial loop	A producer outage becoming a burst of writes
Approval plus live re-read	Acting on a stale incident list
Post-write subscription read	Treating a green command as recovered work

How it works

read_subscriptions (io.kestra.plugin.core.http.Request) reads CREATED Camunda message subscriptions for the exact process and message.
read_instances (io.kestra.plugin.core.http.Request) reads ACTIVE instances older than the SLA. read_outbox (io.kestra.plugin.jdbc.postgresql.Query) reads only old PENDING rows whose published_at is null.
classify intersects the native reads on process instance key and correlation key. The plan includes the event ID and JSON variables and is logged for review.
With auto_remediate: true and dry_run: false, approval pauses. reread_subscriptions, reread_instances and reread_outbox rebuild the plan after resume. stale_plan_guard fails if its stable signature changed.
republish (io.kestra.plugin.core.flow.Loop, concurrencyLimit: 1) calls io.kestra.plugin.camunda.PublishMessage once per candidate with the event ID as messageId and PT0S TTL.
verify_subscriptions and verify_instances read the engine again. fail_still_parked fails if any matching CREATED subscription remains. A Camunda Trigger can invoke the same flow for a repair service task; complete_job (io.kestra.plugin.camunda.CompleteJob) reports success only after verification.
errors posts a Slack alert with the execution ID whenever detection, approval, writing or verification fails.

What you get

A dry-run incident packet showing exact process instance keys, message names, correlation keys and outbox event IDs.
A bounded approval path that republishes only a live, revalidated set.
A native Camunda message command with serial concurrency and deduplication IDs.
A post-write proof from Camunda's own subscription search, plus a failed execution when recovery did not land.

Who it's for

Platform teams operating Camunda 8 order, payment or fulfillment processes with a transactional outbox and a broker handoff. It is for an incident where the producer committed an event but the broker or worker path lost it while the process instance remained at a message catch event.

Why orchestrate this with Kestra

A SQL script can find an old outbox row and a Camunda command can publish a message, but neither one provides an approval snapshot, serial bounded fan-out, stale-plan refusal, task lineage and an alert when the instance stays parked. Kestra keeps the detector, the typed Camunda write and the independent recovery read in one auditable execution.

Tested end to end

Kestra 2.0.5, Camunda 8.9.19 (camunda/camunda@sha256:174f43860c96f25e259785b0d65bc245fbcb20e3d8ff39eef8ff01ae888813ed, one broker and one partition with RDBMS secondary storage), PostgreSQL 16.5 (postgres:16-alpine), plugin-camunda 1.0.3, plugin-jdbc-postgres 2.1.0 and plugin-slack 2.1.0. The fixture deployed a message-catch process and a separate protected process, inserted one old pending outbox row and dropped its broker delivery. Every result below was read from Camunda REST v2 and PostgreSQL with independent probes.

Case	Execution	Result
Healthy control	`19XPXqqe4GRTcH6Tv33iWY`	0 candidates; instance stayed ACTIVE with a CREATED subscription
Degraded dry run	`1lk4L7hhs3AXQazPnc8F14`	1 parked instance and 1 matching old outbox row shown; 0 publishes
Declined approval	`3l26kE9uxzDQoabzvp2Lqg`	1 candidate; approval false; 0 publishes; instance stayed parked
Remediation	`4piUFoc9J09PGIa6n8dFmV`	1 PublishMessage; target subscription became CORRELATED; instance COMPLETED
Repeat	`7iCpvPPLp4DHjjGG8rqLRZ`	0 candidates after recovery; 0 duplicate messages
Ineffective write	`3z99G68xoO8pmjXXlz8XUZ`	Malformed variables caused FAILED; instance stayed ACTIVE with CREATED subscription
Race or partial	`3GeYanS6t96t1FX94mMazr`	Outbox changed during approval; stale signature failed before publish

The exact execution IDs, request payloads, engine snapshots and measured counts are in the evidence packet for this blueprint.

Prerequisites

Camunda 8.9.19 with REST v2 process and message-subscription search, publish-message permission and a reachable Docker network.
PostgreSQL 16 with an outbox_events table containing event_id, message_name, correlation_key, payload, status, published_at and created_at.
Kestra 2.0.5 with Camunda, PostgreSQL JDBC, Slack and Docker runner plugins.
Minimum permissions: read process instances and message subscriptions, publish messages and activate or complete the optional repair job; PostgreSQL SELECT on outbox_events.

Secrets

SLACK_WEBHOOK_URL: required by the error alert.
Use secret-backed Camunda and PostgreSQL credentials in production.

Inputs

Input	Default	Purpose
`camunda_rest_address`	`http://localhost:8080`	Camunda REST base
`outbox_jdbc_url`	`jdbc:postgresql://localhost:5432/outbox`	Producer outbox
`process_definition_id`	`order_fulfillment`	Exact process allowlist
`message_name`	`order-payment-authorized`	Exact message allowlist
`sla_minutes`	`15`	Age floor for both records
`max_candidates`	`10`	Refusal ceiling
`auto_remediate`	`false`	Explicit write permission
`dry_run`	`true`	Explicit safe mode

Quick start

Start Camunda and PostgreSQL on a reachable Docker network and create the outbox_events table using the schema required above.
Add SLACK_WEBHOOK_URL and production connection secrets in Kestra.
Import this flow into namespace company.team and run it with the defaults for a dry run.
Review the candidate keys and event IDs in log_plan. Set auto_remediate: true and dry_run: false only for the approved incident, then answer the Pause with approved: true.
Require a SUCCESS execution and an independent Camunda read showing zero matching CREATED subscriptions before closing the incident.

How to extend

Add another process and message only as a separately reviewed allowlist entry. If an outbox schema stores variables in columns rather than JSON, change the single projection in read_outbox and preserve the stable event ID. For a large backlog, shard by process or incident window and keep each run below its ceiling. Keep the trigger disabled until its job type is deployed with a dedicated worker contract.

Limits: the flow cannot prove that the original producer did not already publish an event outside the outbox's published_at marker and it does not mark the outbox row as published. Operators must reconcile duplicate business effects with the producer's idempotency policy. Camunda secondary search is eventually consistent, so the post-write read waits three seconds and a failed verification remains an incident.

Backfill Camunda Messages Lost Between an Outbox and a Catch Event

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Backfill Camunda Messages Lost Between an Outbox and a Catch Event

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Infrastructure Blueprints

Carbon-Aware Batch Job Executor Companion Flow

Find and revoke over-broad Cassandra permission grants

Casdoor IAM Auth Engine and OIDC Discovery Sentinel

Carbon-Aware Batch Job Scheduler with Deadline-Safe Deferral

Cancel Camunda Process Instances Stuck on an Unresolvable Incident

Cassandra Missing Default TTL Growth Guard

New to Kestra?