Service Bus Dead-Letter Queue Backlog Gate with Kestra

id: servicebus-dead-letter-queue-backlog-gate
namespace: company.team

description: |
  Peek an Azure Service Bus queue's dead-letter sub-queue, classify the backlog HEALTHY,
  WARNING, CRITICAL or PROBE_FAILED from the real count of messages found, alert Slack with
  that count, and log a final audit summary either way.

inputs:
  - id: queue_name
    type: STRING
    displayName: Service Bus queue name
    description: Name of the queue whose dead-letter sub-queue is audited. Maps to
      the Consume task's queueName field.
    defaults: "queue.1"

  - id: max_messages
    type: INT
    displayName: Max messages per probe
    description: Maximum dead-lettered messages pulled in one audit pass. Maps to
      the Consume task's maxMessages field.
    defaults: 50

  - id: max_receive_duration
    type: STRING
    displayName: Max receive duration
    description: ISO-8601 duration the probe waits for messages before returning.
      Maps to the Consume task's maxReceiveDuration field.
    defaults: "PT10S"

  - id: violation_count_critical_threshold
    type: INT
    displayName: Violation count threshold
    description: Number of dead-lettered messages found at or above which the run is
      classified at violation_severity instead of HEALTHY.
    defaults: 1

  - id: violation_severity
    type: SELECT
    displayName: Violation severity
    description: How a non-empty dead-letter queue is classified. critical pages
      immediately, warning logs it as a lower-urgency finding for a queue with a
      known, tolerated dead-letter rate.
    values:
      - "critical"
      - "warning"
    defaults: "critical"

  - id: use_emulator
    type: BOOL
    displayName: Use local Service Bus emulator
    description: When true, connects with the official emulator's fixed connection
      string. When false, connects with a real Service Bus connection string
      from a secret.
    defaults: true

  - id: emulator_connection_string
    type: STRING
    displayName: Emulator connection string
    description: Only used when use_emulator is true. The Service Bus emulator's own
      documented fixed connection string, not a real secret.
    defaults: "Endpoint=sb://localhost;SharedAccessKeyName=RootManageSharedAccessKe\
      y;SharedAccessKey=SAS_KEY_VALUE;UseDevelopmentEmulator=true;"

tasks:
  - id: route_connection_mode
    type: io.kestra.plugin.core.flow.If
    description: Picks the connection. The emulator branch uses the fixed emulator
      connection string, the live branch uses a real Service Bus connection
      string from a secret. Both also set tenantId, a field Consume inherits
      from AbstractAzureIdentityConnection and marks @NotNull unconditionally
      even though applyAuth() only reads it when connectionString is absent, so
      it is required by schema but inert here.
    condition: "{{ inputs.use_emulator }}"
    then:
      - id: probe_dlq_emulator
        type: io.kestra.plugin.azure.servicebus.Consume
        description: Peeks the dead-letter sub-queue. PEEK_LOCK is the task's own
          default and it never calls complete() on a message, so every peeked
          message returns to the dead-letter queue once its lock expires, this
          does not drain the backlog. tenantId is a placeholder, not a real
          Azure AD tenant, see the flow-level description above.
        allowFailure: true
        connectionString: "{{ inputs.emulator_connection_string }}"
        tenantId: "00000000-0000-0000-0000-000000000000"
        queueName: "{{ inputs.queue_name }}"
        subQueue: DEAD_LETTER_QUEUE
        maxMessages: "{{ inputs.max_messages }}"
        maxReceiveDuration: "{{ inputs.max_receive_duration }}"
    else:
      - id: probe_dlq_live
        type: io.kestra.plugin.azure.servicebus.Consume
        description: Same dead-letter sub-queue peek against real Service Bus. tenantId
          is a placeholder, not a real Azure AD tenant, see the flow-level
          description above.
        allowFailure: true
        connectionString: "{{ secret('SERVICEBUS_CONNECTION_STRING') }}"
        tenantId: "00000000-0000-0000-0000-000000000000"
        queueName: "{{ inputs.queue_name }}"
        subQueue: DEAD_LETTER_QUEUE
        maxMessages: "{{ inputs.max_messages }}"
        maxReceiveDuration: "{{ inputs.max_receive_duration }}"

  - id: classify_dlq_status
    type: io.kestra.plugin.core.debug.Return
    description: PROBE_FAILED when neither probe branch returned at all,
      violation_severity when the real count meets the threshold, HEALTHY
      otherwise.
    format: "{{ (outputs.probe_dlq_emulator is not defined and
      outputs.probe_dlq_live is not defined) ? 'PROBE_FAILED' :
      (((outputs.probe_dlq_emulator is defined ?
      outputs.probe_dlq_emulator.count : outputs.probe_dlq_live.count) >=
      inputs.violation_count_critical_threshold) ? (inputs.violation_severity ==
      'critical' ? 'CRITICAL' : 'WARNING') : 'HEALTHY') }}"

  - id: route_by_dlq_status
    type: io.kestra.plugin.core.flow.Switch
    description: Branch A logs an empty dead-letter queue. Branch B alerts Slack
      with the real message count when the dead-letter queue is not empty, or
      the probe itself failed.
    value: "{{ outputs.classify_dlq_status.value }}"
    cases:
      "HEALTHY":
        - id: log_healthy
          type: io.kestra.plugin.core.log.Log
          message: "Service Bus queue {{ inputs.queue_name }} dead-letter sub-queue empty
            over this probe."
      "WARNING":
        - id: alert_warning
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {{ {"text": "Service Bus queue " ~ inputs.queue_name ~ " dead-letter sub-queue WARNING. " ~ (outputs.probe_dlq_emulator is defined ? outputs.probe_dlq_emulator.count : outputs.probe_dlq_live.count) ~ " message(s) found."} | toJson }}
      "CRITICAL":
        - id: alert_critical
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {{ {"text": "Service Bus queue " ~ inputs.queue_name ~ " dead-letter sub-queue CRITICAL. " ~ (outputs.probe_dlq_emulator is defined ? outputs.probe_dlq_emulator.count : outputs.probe_dlq_live.count) ~ " message(s) found, at or above the " ~ inputs.violation_count_critical_threshold ~ " threshold."} | toJson }}
      "PROBE_FAILED":
        - id: alert_probe_failed
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {{ {"text": "Service Bus dead-letter probe for queue " ~ inputs.queue_name ~ " did not return a result. Check queue_name, use_emulator, and the Service Bus connection string."} | toJson }}
    defaults:
      - id: log_unexpected_status
        type: io.kestra.plugin.core.log.Log
        message: "Unexpected classify_dlq_status value: {{
          outputs.classify_dlq_status.value }}"

  - id: log_audit_summary
    type: io.kestra.plugin.core.log.Log
    description: Final audit log confirming the gate ran, regardless of the branch
      taken above.
    message: "Service Bus dead-letter queue backlog gate completed. queue={{
      inputs.queue_name }} status={{ outputs.classify_dlq_status.value }}
      use_emulator={{ inputs.use_emulator }}"

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: |
      {{ {"text": "Service Bus dead-letter queue backlog gate flow failed (execution " ~ execution.id ~ "). Check the execution logs."} | toJson }}

triggers:
  - id: every_fifteen_minutes
    type: io.kestra.plugin.core.trigger.Schedule
    description: Shipped disabled, and always against the local emulator by default.
    cron: "*/15 * * * *"
    disabled: true
    inputs:
      queue_name: "queue.1"
      max_messages: 50
      max_receive_duration: "PT10S"
      violation_count_critical_threshold: 1
      violation_severity: "critical"
      use_emulator: true
      emulator_connection_string: "Endpoint=sb://localhost;SharedAccessKeyName=RootMa\
        nageSharedAccessKey;SharedAccessKey=SAS_KEY_VALUE;UseDevelopmentEmulato\
        r=true;"

A message that exhausts MaxDeliveryCount or expires past DefaultMessageTimeToLive does not vanish, Service Bus moves it to the queue's own dead-letter sub-queue and leaves it there, invisible to anything reading the main queue, until someone happens to look. A growing dead-letter queue means messages a downstream consumer could not process, exactly the kind of silent failure that only shows up in a postmortem. This blueprint peeks the dead-letter sub-queue on a schedule, counts what is actually there, and alerts Slack with the real number, without draining or altering a single message.

How it works

route_connection_mode (io.kestra.plugin.core.flow.If) picks the connection based on use_emulator. Both branches run io.kestra.plugin.azure.servicebus.Consume against queue_name with subQueue: DEAD_LETTER_QUEUE, with allowFailure: true so a bad credential or a renamed queue does not stop the flow before it can be classified as PROBE_FAILED below. Consume defaults to PEEK_LOCK receive mode and never calls complete() on a message, so every message it reads returns to the dead-letter sub-queue once its lock expires, this probe never drains the backlog it is reporting on. Both branches also set tenantId to a placeholder GUID: Consume inherits this field from AbstractAzureIdentityConnection (in the separate plugin-azure-lib repository), which marks it @NotNull unconditionally, even though applyAuth() only reads tenantId when connectionString is absent, so it is required by schema but never actually used here.
classify_dlq_status (io.kestra.plugin.core.debug.Return) reads whichever branch's count exists (guarded with is defined/is not defined, since exactly one branch ran) and returns PROBE_FAILED when neither branch returned at all, violation_severity (CRITICAL or WARNING) when the count meets violation_count_critical_threshold, HEALTHY otherwise.
route_by_dlq_status (io.kestra.plugin.core.flow.Switch) branches on that status: HEALTHY logs it, WARNING, CRITICAL, and PROBE_FAILED each alert Slack with the real count that triggered it, never an invented number.
log_audit_summary always runs last and logs the final classification.
The errors block posts to Slack if the flow itself fails outside the gate logic above.

What you get

A real count of what is actually sitting in a dead-letter sub-queue, not a guess from whether the main queue still looks busy.
A Slack alert carrying the exact count that triggered it, never an invented number.
A genuinely non-destructive probe: PEEK_LOCK with no complete() call means the dead-letter queue is exactly as full after this flow runs as before.
A fully local test path against Microsoft's own official Service Bus emulator, no real Azure namespace needed.

Who it's for

Messaging and platform teams running Azure Service Bus who need to know about a growing dead-letter backlog before a downstream consumer's silent failures turn into a postmortem.
Teams evaluating the Service Bus plugin who want a working example of Consume against the DEAD_LETTER_QUEUE sub-queue, with both the emulator connection string and a real Service Bus connection string.

Why orchestrate this with Kestra

Service Bus itself has no built-in alert for "the dead-letter queue is not empty", that requires wiring a separate Azure Monitor alert rule on a metric. Kestra adds the schedule, the threshold and severity policy, and a Slack alert built from the real count Consume itself returns, all as one declarative YAML file you can version and review like code, runnable against the official emulator before it ever touches a real namespace.

Prerequisites

For local testing, the official Azure Service Bus emulator, no Azure account needed. Save this as config.json:

{
  "UserConfig": {
    "Namespaces": [
      {
        "Name": "sbemulatorns",
        "Queues": [
          {
            "Name": "queue.1",
            "Properties": {
              "DeadLetteringOnMessageExpiration": false,
              "DefaultMessageTimeToLive": "PT1H",
              "LockDuration": "PT1M",
              "MaxDeliveryCount": 3,
              "RequiresDuplicateDetection": false,
              "RequiresSession": false
            }
          }
        ]
      }
    ],
    "Logging": { "Type": "File" }
  }
}

Then run, with CONFIG_PATH pointed at that file and ACCEPT_EULA=Y after reading the emulator's own EULA:

docker run -d --name servicebus-emulator -p 5672:5672 -p 5300:5300 \\
  -v "$CONFIG_PATH:/ServiceBus_Emulator/ConfigFiles/Config.json" \\
  -e ACCEPT_EULA=Y -e MSSQL_SA_PASSWORD="<a-strong-password>" -e SQL_SERVER=mssql \\
  mcr.microsoft.com/azure-messaging/servicebus-emulator:latest

The emulator also needs its SQL Server Linux dependency reachable as mssql, the official docs' docker-compose.yaml wires this correctly, a bare docker run of the emulator alone is not enough, see the Links section. Port 5300 does not collide with Kestra's own default UI port 8080.

Messages only reach the dead-letter sub-queue when something actually dead-letters them, exceeding MaxDeliveryCount on a locked, never-completed message is the simplest way to produce one for testing.
For real Service Bus, a connection string with Listen claim on the queue (a shared access policy's primary or secondary key from the Azure portal or az servicebus queue authorization-rule keys list). Consume also exposes tenantId/clientId/ clientSecret/pemCertificate for Azure AD service-principal auth (inherited from AbstractAzureIdentityConnection), unused by this blueprint since both branches authenticate with connectionString instead.
A Slack incoming webhook for alerts.

Secrets

SERVICEBUS_CONNECTION_STRING: used by probe_dlq_live only, not needed when use_emulator is true.
SLACK_WEBHOOK: Slack incoming webhook URL used by the alert tasks and the errors block.

In Kestra OSS, set these as base64-encoded environment variables, for example SECRET_SERVICEBUS_CONNECTION_STRING, SECRET_SLACK_WEBHOOK.

Inputs

Input	Type	Default	Allowed values
`queue_name`	STRING	`queue.1`	any Service Bus queue name
`max_messages`	INT	`50`	any positive integer
`max_receive_duration`	STRING	`PT10S`	any ISO-8601 duration
`violation_count_critical_threshold`	INT	`1`	any non-negative integer
`violation_severity`	SELECT	`critical`	`critical`, `warning`
`use_emulator`	BOOL	`true`	`true`, `false`
`emulator_connection_string`	STRING	see description	the emulator's own fixed connection string

Quick start

Start the Service Bus emulator with the config.json above.
Send a few messages to queue.1 that are guaranteed to dead-letter, for example by receiving and abandoning the same message more than MaxDeliveryCount times.
Add the SLACK_WEBHOOK secret.
Run this flow manually with the defaults, it should report CRITICAL.
Set use_emulator: false and add the SERVICEBUS_CONNECTION_STRING secret to point this at a real queue.
Enable the every_fifteen_minutes trigger.

How to extend

Add a second Consume with topicName/subscriptionName instead of queueName to audit a topic subscription's dead-letter sub-queue the same way, queues and subscriptions both have one.
Loop this flow over a list of queue names with io.kestra.plugin.core.flow.Loop to audit every queue in a namespace from one flow.
Read each message's DeadLetterReason and DeadLetterErrorDescription application properties, included in the stored message body, to group the Slack alert by why messages are dead-lettering, not only how many.

Pitfalls

count reflects messages this probe actually pulled within max_messages and max_receive_duration, not necessarily the sub-queue's total depth if it holds more than max_messages, raise both for a more complete count on a very backed-up queue.
PEEK_LOCK still locks each peeked message for its LockDuration, other consumers cannot process it during that window, a very short LockDuration combined with a slow max_receive_duration can briefly starve a legitimate dead-letter reprocessing job running at the same time.
This blueprint never completes, abandons, or deletes a dead-lettered message, there is no remediation step, deciding whether a dead-lettered message is safe to replay back to the main queue needs a human to look at why it landed there first.
The emulator config's DeadLetteringOnMessageExpiration: false in the sample above does not disable dead-lettering entirely, MaxDeliveryCount: 3 still dead-letters a message that exhausts its delivery attempts.
tenantId is set to a placeholder GUID on both Consume tasks and is not a real Azure AD tenant. AbstractAzureIdentityConnection marks it @NotNull unconditionally, so a flow using connectionString auth still fails schema validation without it, even though applyAuth() never reads it in that case, verified directly from the plugin's source.

Service Bus Dead-Letter Queue Backlog Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Quick start

How to extend

Pitfalls

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Service Bus Dead-Letter Queue Backlog Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Quick start

How to extend

Pitfalls

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Infrastructure Blueprints

Prune noncurrent object versions billed forever behind a versioned bucket

Memcached Slab Eviction Sentinel

Meilisearch Indexing Queue Staleness Sentinel

Incremental, idempotent lake ingestion from MinIO with DuckDB

Catch MinIO Replication Drift Before Objects Go Missing on the Target

Scheduled MinIO Bucket Retention Cleanup

More Cloud Blueprints

Huawei DataArts Factory Job Run Health Gate

Audit HTTP Security Headers Across URLs

Huawei Cloud Eye (CES) Metric Threshold Health Gate

Huawei DLI Data Quality Assertion Gate

Idempotent Incremental CSV Loader from S3 to PostgreSQL

Huawei OBS Stale Object Retention Sentinel

New to Kestra?