Solace DLQ Backlog Drain Gate for Kestra

id: solace-dlq-backlog-drain-gate
namespace: company.team
description: |
  Drain a sample of messages from a Solace PubSub+ dead-letter or quarantine
  queue on a schedule, classify the result as BACKLOGGED, HEALTHY, or
  PROBE_FAILED based on how many messages came off in the sample window,
  log clean passes, and alert Slack the moment the queue is backlogged or
  unreachable.

inputs:
  - id: solace_host
    type: STRING
    defaults: "localhost:55555"
    displayName: "Solace Host"
    description: "Broker hostname and port for the Solace PubSub+ event broker, for
      example localhost:55555."

  - id: solace_vpn
    type: STRING
    defaults: "default"
    displayName: "Solace Message VPN"
    description: "Message VPN name to connect to on the Solace broker."

  - id: dlq_queue_name
    type: STRING
    defaults: "audit.dlq.queue"
    displayName: "Dead-Letter Queue Name"
    description: "Name of the Solace dead-letter or quarantine queue to drain and
      audit for backlog accumulation."

  - id: queue_type
    type: SELECT
    values:
      - "DURABLE_EXCLUSIVE"
      - "DURABLE_NON_EXCLUSIVE"
      - "NON_DURABLE_EXCLUSIVE"
    defaults: "DURABLE_NON_EXCLUSIVE"
    displayName: "Queue Type"
    description: "Solace queue access type used when binding the consumer to the
      dead-letter queue. Must match how the queue was provisioned on the
      broker."

  - id: message_deserializer
    type: SELECT
    values:
      - "STRING"
      - "BINARY"
      - "ION"
      - "JSON"
    defaults: "STRING"
    displayName: "Message Deserializer"
    description: "Format used to deserialize drained messages before writing the
      batch to internal storage."

  - id: max_messages_per_drain
    type: INT
    defaults: 100
    displayName: "Max Messages Per Drain"
    description: "Maximum number of messages to pull from the dead-letter queue in a
      single audit pass."

  - id: max_drain_duration
    type: DURATION
    defaults: PT10S
    displayName: "Max Drain Duration"
    description: "Maximum time to keep polling the dead-letter queue during a single
      audit pass, expressed as an ISO-8601 duration."

  - id: backlog_alert_threshold
    type: INT
    defaults: 25
    displayName: "Backlog Alert Threshold"
    description: "Number of drained messages at or above which the dead-letter queue
      is classified as backlogged and an alert is dispatched."

  - id: send_slack_alert
    type: STRING
    defaults: "true"
    displayName: "Send Slack Alert"
    description: "When set to true, dispatch a Slack notification on a backlogged
      classification; when false, only log the breach."

triggers:
  - id: solace_dlq_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    description: "Drain and audit the dead-letter queue every 6 hours."
    cron: "0 */6 * * *"
    disabled: true
    inputs:
      solace_host: "localhost:55555"
      solace_vpn: "default"
      dlq_queue_name: "audit.dlq.queue"
      queue_type: "DURABLE_NON_EXCLUSIVE"
      message_deserializer: "STRING"
      max_messages_per_drain: 100
      max_drain_duration: PT10S
      backlog_alert_threshold: 25
      send_slack_alert: "true"

tasks:
  - id: drain_dlq_sample
    type: io.kestra.plugin.solace.Consume
    description: "Drain up to max_messages_per_drain messages, or stop after
      max_drain_duration, from the dead-letter queue and write the batch to
      internal storage."
    host: "{{ inputs.solace_host }}"
    username: "{{ secret('SOLACE_USERNAME') }}"
    password: "{{ secret('SOLACE_PASSWORD') }}"
    vpn: "{{ inputs.solace_vpn }}"
    queueName: "{{ inputs.dlq_queue_name }}"
    queueType: "{{ inputs.queue_type }}"
    messageDeserializer: "{{ inputs.message_deserializer }}"
    maxMessages: "{{ inputs.max_messages_per_drain }}"
    maxDuration: "{{ inputs.max_drain_duration }}"
    allowFailure: true

  - id: classify_backlog_status
    type: io.kestra.plugin.core.debug.Return
    description: "Classify as PROBE_FAILED when the drain produced no output (broker
      unreachable or bind failure), BACKLOGGED when the drained count is at or
      above backlog_alert_threshold, otherwise HEALTHY."
    format: "{{ (outputs.drain_dlq_sample.messagesCount is defined) ?
      ((outputs.drain_dlq_sample.messagesCount >=
      inputs.backlog_alert_threshold) ? 'BACKLOGGED' : 'HEALTHY') :
      'PROBE_FAILED' }}"

  - id: route_by_backlog_status
    type: io.kestra.plugin.core.flow.Switch
    description: "Branch on the classification: page Slack when backlogged or
      unreachable, otherwise log a healthy audit line."
    value: "{{ outputs.classify_backlog_status.value }}"
    cases:
      "BACKLOGGED":
        - id: evaluate_alert_toggle
          type: io.kestra.plugin.core.flow.If
          description: "Only call the Slack webhook when send_slack_alert is true;
            otherwise log the breach without paging."
          condition: "{{ inputs.send_slack_alert == 'true' }}"
          then:
            - id: dispatch_backlog_alert
              type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
              description: "Page the messaging/SRE channel with the drained count and
                threshold."
              url: "{{ secret('SLACK_WEBHOOK') }}"
              payload: |
                {
                  "text": "Solace DLQ backlog alert: queue '{{ inputs.dlq_queue_name }}' drained {{ outputs.drain_dlq_sample.messagesCount }} messages in this pass, at or above threshold {{ inputs.backlog_alert_threshold }}. VPN {{ inputs.solace_vpn }}, host {{ inputs.solace_host }}. Execution {{ execution.id }}."
                }
          else:
            - id: log_alert_suppressed
              type: io.kestra.plugin.core.log.Log
              description: "Record the breach even when Slack paging is disabled, so the
                execution history still shows it."
              message: "Backlog threshold breached for queue {{ inputs.dlq_queue_name }} ({{
                outputs.drain_dlq_sample.messagesCount }} messages) but Slack
                alert suppressed by send_slack_alert=false."
      "HEALTHY":
        - id: log_queue_healthy
          type: io.kestra.plugin.core.log.Log
          description: "Record a healthy audit result when drained count is below
            threshold."
          message: "Solace dead-letter queue {{ inputs.dlq_queue_name }} is healthy.
            Drained {{ outputs.drain_dlq_sample.messagesCount }} messages
            (threshold {{ inputs.backlog_alert_threshold }})."
      "PROBE_FAILED":
        - id: alert_probe_failure
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: "Page the channel when the drain itself could not run, since the
            backlog state is now unknown."
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {
              "text": "Solace DLQ sentinel could not drain queue '{{ inputs.dlq_queue_name }}' on host {{ inputs.solace_host }}, VPN {{ inputs.solace_vpn }}. Check broker connectivity, credentials, and queue_type. Execution {{ execution.id }}."
            }
    defaults:
      - id: log_unclassified
        type: io.kestra.plugin.core.log.Log
        description: "Fallback branch; reached only if classify_backlog_status produces
          a value outside the three known cases."
        message: "Solace DLQ sentinel produced an unclassified status for queue {{
          inputs.dlq_queue_name }}: {{ outputs.classify_backlog_status.value
          }}."

  - id: log_audit_summary
    type: io.kestra.plugin.core.log.Log
    description: "Final post-audit status line recorded on every run regardless of branch."
    message: "Solace DLQ backlog audit completed for queue '{{ inputs.dlq_queue_name
      }}'. Status: {{ outputs.classify_backlog_status.value }}. Execution ID: {{
      execution.id }}."

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: "Flow-level error handler: page the channel when the sentinel flow
      itself fails outright."
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: |
      {
        "text": "CRITICAL: Solace DLQ backlog sentinel flow failed during audit of queue '{{ inputs.dlq_queue_name }}' on host {{ inputs.solace_host }}. Execution ID: {{ execution.id }}."
      }

outputs:
  - id: status
    type: STRING
    value: "{{ outputs.classify_backlog_status.value }}"

A Solace dead-letter queue that quietly fills up is invisible until a downstream team notices messages went missing. plugin-solace has no dedicated queue-depth or backlog metric task, so this blueprint uses its native Consume task as the probe itself: it drains a bounded sample of messages from the dead-letter queue, counts how many came off within the sample window, and classifies the result as BACKLOGGED, HEALTHY, or PROBE_FAILED before deciding whether to page Slack.

How it works

drain_dlq_sample (io.kestra.plugin.solace.Consume) binds to dlq_queue_name as queue_type and pulls up to max_messages_per_drain messages within max_drain_duration, writing the deserialized batch to internal storage. allowFailure: true lets the flow continue instead of aborting if the broker is unreachable or the bind fails.
classify_backlog_status (io.kestra.plugin.core.debug.Return) checks whether outputs.drain_dlq_sample.messagesCount is defined at all. If it is missing, the classification is PROBE_FAILED. If it is defined and at or above backlog_alert_threshold, the classification is BACKLOGGED; otherwise HEALTHY.
route_by_backlog_status (io.kestra.plugin.core.flow.Switch) branches on that string. BACKLOGGED runs evaluate_alert_toggle (io.kestra.plugin.core.flow.If), which pages Slack via dispatch_backlog_alert only when send_slack_alert is true, logging a suppressed-alert line otherwise. HEALTHY logs a clean audit line. PROBE_FAILED pages Slack directly since the backlog state is now unknown. The defaults branch logs a fallback line that is unreachable under normal operation.
log_audit_summary records a status line on every run regardless of branch, so execution history reads as a continuous backlog trend.
The flow-level errors block pages Slack separately if the sentinel flow itself fails outright.
The solace_dlq_schedule trigger polls every 6 hours with explicit input overrides, shipped disabled: true until pointed at a real broker and queue.

What you get

A scheduled, bounded drain of a Solace dead-letter queue with zero custom consumer code.
A three-way status output (BACKLOGGED / HEALTHY / PROBE_FAILED) other flows can gate on.
A Slack page naming the queue, drained count, threshold, VPN, and host on breach or probe failure.
An internal-storage URI (outputs.drain_dlq_sample.uri) containing the drained message batch for manual inspection.
A separate failure channel for the sentinel flow itself, plus a standing per-run audit log line.

Who it's for

Messaging platform engineers and SREs operating Solace PubSub+ event brokers in production.
Data and integration teams whose pipelines route failed deliveries to a Solace dead-letter or quarantine queue and need to know when it backs up.
Platform teams standardizing Day 2 backlog governance across message-bus-backed services inside Kestra.

Why orchestrate this with Kestra

A one-off CLI drain or a custom cron script proves the queue depth for one instant and then the evidence is gone. This flow keeps every drain in Kestra's execution history, turns the result into a typed status output other flows can depend on, stores the drained batch in internal storage for inspection, and gives you an If/Switch branch to grow into real remediation (reprocessing, ticketing) without rewriting the probe itself.

Prerequisites

A running Solace PubSub+ broker reachable from the Kestra Worker, with a dead-letter or quarantine queue already provisioned as the access type you select in queue_type.
Broker credentials with consume permission on that queue.
A Slack incoming webhook for the backlog, probe-failure, and flow-failure alerts.

Quick start

Add the SOLACE_USERNAME, SOLACE_PASSWORD, and SLACK_WEBHOOK secrets.
Point dlq_queue_name and queue_type at a real dead-letter queue on your broker (see Pitfalls before pointing this at a live production queue).
Run the flow once and confirm status reads HEALTHY or BACKLOGGED as expected for your current queue depth.
Enable solace_dlq_schedule.

How to extend

Parse the drained batch at outputs.drain_dlq_sample.uri to summarize failure reasons by message property instead of just counting.
Push status into a KV store to alert only on state transitions instead of every run.
Fan the probe out to several dead-letter queues with a Loop task to monitor multiple services from one flow.
Chain a io.kestra.plugin.solace.Produce task to republish selectively reprocessed messages back onto their source topic after manual review.

Secrets

SOLACE_USERNAME: Broker username with consume permission on the dead-letter queue.
SOLACE_PASSWORD: Broker password for that user.
SLACK_WEBHOOK: Slack incoming webhook used for the backlog, probe-failure, and flow-failure alerts.
OSS note: if your Kestra instance reads secrets from environment variables instead of a secrets backend, set SECRET_SOLACE_USERNAME, SECRET_SOLACE_PASSWORD, and SECRET_SLACK_WEBHOOK as base64, for example echo -n "<your-value>" | base64.

Inputs

solace_host (STRING, default localhost:55555): broker hostname and port.
solace_vpn (STRING, default default): Solace Message VPN name.
dlq_queue_name (STRING, default audit.dlq.queue): dead-letter queue to drain.
queue_type (SELECT, one of DURABLE_EXCLUSIVE, DURABLE_NON_EXCLUSIVE, NON_DURABLE_EXCLUSIVE, default DURABLE_NON_EXCLUSIVE): queue access type as provisioned on the broker.
message_deserializer (SELECT, one of STRING, BINARY, ION, JSON, default STRING): deserializer applied to drained messages.
max_messages_per_drain (INT, default 100): maximum messages pulled per audit pass.
max_drain_duration (DURATION, default PT10S): maximum polling time per audit pass.
backlog_alert_threshold (INT, default 25): drained count at or above which the queue is classified BACKLOGGED.
send_slack_alert (STRING, default "true"): when "true", page Slack on a BACKLOGGED classification; when "false", only log it.

Outputs

{{ outputs.drain_dlq_sample.messagesCount }}: number of messages drained in this pass.
{{ outputs.drain_dlq_sample.uri }}: internal storage URI (kestra://) containing the drained message batch.
{{ outputs.classify_backlog_status.value }}: classification string, one of BACKLOGGED, HEALTHY, PROBE_FAILED.
{{ outputs.status }}: flow-level alias of the same classification.

Pitfalls

Consuming from a durable Solace queue acknowledges and permanently removes the drained messages; this is not a non-destructive peek. Point dlq_queue_name only at a dedicated dead-letter or quarantine queue, never a live production work queue.
messagesCount is a bounded point-in-time sample limited by both max_messages_per_drain and max_drain_duration; a fast upstream failure can refill the queue between runs and mask sustained backlog.
queue_type must match how the queue was actually provisioned on the broker; a mismatch causes a bind failure, which this flow classifies as PROBE_FAILED rather than HEALTHY.
The Solace PubSub+ software broker's management UI defaults to host port 8080, the same port Kestra's own webserver commonly uses; the local container setup below maps it to 8180 instead.
The vpn must already exist on the broker; only a freshly initialized broker has a Message VPN literally named default.

Local Container Setup

To start a local Solace PubSub+ Standard broker container for testing, execute:

docker run -d --name kestra-solace -p 8180:8080 -p 55555:55555 --shm-size=1g --env username_admin_globalaccesslevel=admin --env username_admin_password=admin --ulimit nofile=2448:1048576 solace/solace-pubsub-standard

If Kestra is running inside a Docker network, connect the broker container to Kestra's network:

docker network connect YOUR_KESTRA_NETWORK kestra-solace

(Run docker network ls to verify your active Kestra network name.)

Solace Dead-Letter Queue Backlog Drain Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Quick start

How to extend

Secrets

Inputs

Outputs

Pitfalls

Local Container Setup

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Solace Dead-Letter Queue Backlog Drain Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Quick start

How to extend

Secrets

Inputs

Outputs

Pitfalls

Local Container Setup

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Infrastructure Blueprints

Audit SNS Fanout Failures Before Subscribers Drift

SLSA Supply-Chain Attestation Admission Gate

Solace Dead-Letter Message Triage & Safe Replay Gate

SMB Share Ingestion Stall and Quarantine Sentinel

Snort Packet Processing Stall and Rule Eval Timeout Sentinel

Solace Event Mesh Synthetic Delivery and Round-Trip Latency Sentinel

More Data Blueprints

Feast Feature Store Offline-to-Online Sync Drift & Parity Guard

Flink Stateful Job Savepoint and Restart Drill

Flink Job Restart Loop Sentinel

FHIR Pending Lab Results Sentinel

Trigger multiple Fivetran syncs in parallel, then run a dbt job

Firecrawl Web Scraping Job Error and Stagnation Sentinel

New to Kestra?