JMS Poison Message Quarantine Gate for Kestra

id: jms-poison-message-quarantine-gate
namespace: company.team
description: |
  Audit a JMS dead-letter destination on a schedule using a server-side
  JMSXDeliveryCount selector to isolate genuinely poisoned messages (those
  redelivered past a threshold) from ordinary dead-letter arrivals,
  classify the result as POISON_QUARANTINE_BREACH, HEALTHY, or PROBE_FAILED,
  log clean passes, and alert Slack the moment poison messages pile up or
  the broker cannot be reached.

inputs:
  - id: jms_broker_url
    type: STRING
    defaults: "tcp://localhost:61616"
    displayName: "JMS Broker URL"
    description: "Broker connection URL applied as the brokerURL connection property
      on the JMS ConnectionFactory, for example tcp://localhost:61616."

  - id: jms_connection_factory_class
    type: STRING
    defaults: "org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory"
    displayName: "JMS ConnectionFactory Class"
    description: "Fully qualified class name of the JMS ConnectionFactory
      implementation to instantiate directly (DIRECT mode)."

  - id: provider_jar_path
    type: STRING
    defaults: "kestra:///jms/artemis-client.jar"
    displayName: "Provider JAR Path"
    description: "Internal storage URI or file path to the JMS provider client JAR.
      If no provider JAR is reachable at this path, drop the matching JAR under
      the plugin's jms-libs folder on the Worker instead and leave this blank."

  - id: dlq_destination_name
    type: STRING
    defaults: "ActiveMQ.DLQ"
    displayName: "Dead-Letter Destination Name"
    description: "Name of the JMS dead-letter queue or topic to audit for poison messages."

  - id: message_serde_type
    type: SELECT
    values:
      - "STRING"
      - "JSON"
      - "BYTES"
    defaults: "STRING"
    displayName: "Message Deserializer"
    description: "Format used to deserialize audited message bodies before writing
      the batch to internal storage."

  - id: poison_redelivery_threshold
    type: INT
    defaults: 3
    displayName: "Poison Redelivery Threshold"
    description: "Minimum JMSXDeliveryCount value used in a server-side message
      selector; messages redelivered more than this many times are treated as
      poison rather than ordinary dead-letter arrivals."

  - id: max_messages_per_audit
    type: INT
    defaults: 50
    displayName: "Max Messages Per Audit"
    description: "Maximum number of matching poison messages to pull from the
      dead-letter destination in a single audit pass."

  - id: max_wait_timeout_ms
    type: INT
    defaults: 5000
    displayName: "Max Wait Timeout (ms)"
    description: "Bounded wait time in milliseconds for the audit poll. plugin-jms
      defaults this to 0, which waits indefinitely, so this input must stay
      above zero to keep the sentinel from hanging on an idle destination."

  - id: poison_alert_threshold
    type: INT
    defaults: 5
    displayName: "Poison Count Alert Threshold"
    description: "Number of matching poison messages at or above which the
      destination is classified as a quarantine breach 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 quarantine
      breach classification; when false, only log the breach."

triggers:
  - id: jms_poison_audit_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    description: "Audit the dead-letter destination for poison messages every 6 hours."
    cron: "0 */6 * * *"
    disabled: true
    inputs:
      jms_broker_url: "tcp://localhost:61616"
      jms_connection_factory_class: "org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory"
      provider_jar_path: "kestra:///jms/artemis-client.jar"
      dlq_destination_name: "ActiveMQ.DLQ"
      message_serde_type: "STRING"
      poison_redelivery_threshold: 3
      max_messages_per_audit: 50
      max_wait_timeout_ms: 5000
      poison_alert_threshold: 5
      send_slack_alert: "true"

tasks:
  - id: audit_poison_messages
    type: io.kestra.plugin.jms.Consume
    description: "Consume only messages whose JMSXDeliveryCount exceeds
      poison_redelivery_threshold from the dead-letter destination, bounded by
      max_messages_per_audit and max_wait_timeout_ms."
    connectionFactoryConfig:
      type: DIRECT
      connectionFactoryClass: "{{ inputs.jms_connection_factory_class }}"
      providerJarPaths: "{{ inputs.provider_jar_path }}"
      username: "{{ secret('JMS_USERNAME') }}"
      password: "{{ secret('JMS_PASSWORD') }}"
      connectionProperties:
        brokerURL: "{{ inputs.jms_broker_url }}"
    destination:
      destinationName: "{{ inputs.dlq_destination_name }}"
      destinationType: QUEUE
    messageSelector: "JMSXDeliveryCount > {{ inputs.poison_redelivery_threshold }}"
    serdeType: "{{ inputs.message_serde_type }}"
    maxMessages: "{{ inputs.max_messages_per_audit }}"
    maxWaitTimeout: "{{ inputs.max_wait_timeout_ms }}"
    allowFailure: true

  - id: classify_quarantine_status
    type: io.kestra.plugin.core.debug.Return
    description: "Classify as PROBE_FAILED when the audit produced no output (broker
      unreachable, missing provider JAR, or bind failure),
      POISON_QUARANTINE_BREACH when the matched poison count is at or above
      poison_alert_threshold, otherwise HEALTHY."
    format: "{{ (outputs.audit_poison_messages.count is defined) ?
      ((outputs.audit_poison_messages.count >= inputs.poison_alert_threshold) ?
      'POISON_QUARANTINE_BREACH' : 'HEALTHY') : 'PROBE_FAILED' }}"

  - id: route_by_quarantine_status
    type: io.kestra.plugin.core.flow.Switch
    description: "Branch on the classification: page Slack when poison messages
      breach threshold or the probe fails, otherwise log a healthy audit line."
    value: "{{ outputs.classify_quarantine_status.value }}"
    cases:
      "POISON_QUARANTINE_BREACH":
        - 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_poison_alert
              type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
              description: "Page the messaging/SRE channel with the poison count and
                threshold."
              url: "{{ secret('SLACK_WEBHOOK') }}"
              payload: |
                {
                  "text": "JMS poison message quarantine breach: destination '{{ inputs.dlq_destination_name }}' matched {{ outputs.audit_poison_messages.count }} messages with JMSXDeliveryCount over {{ inputs.poison_redelivery_threshold }}, at or above alert threshold {{ inputs.poison_alert_threshold }}. Broker {{ inputs.jms_broker_url }}. 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: "Poison quarantine threshold breached for destination {{
                inputs.dlq_destination_name }} ({{
                outputs.audit_poison_messages.count }} messages) but Slack alert
                suppressed by send_slack_alert=false."
      "HEALTHY":
        - id: log_destination_healthy
          type: io.kestra.plugin.core.log.Log
          description: "Record a healthy audit result when matched poison count is below
            threshold."
          message: "JMS dead-letter destination {{ inputs.dlq_destination_name }} is
            healthy. Matched {{ outputs.audit_poison_messages.count }} poison
            messages (threshold {{ inputs.poison_alert_threshold }})."
      "PROBE_FAILED":
        - id: alert_probe_failure
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: "Page the channel when the audit itself could not run, since the
            quarantine state is now unknown."
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {
              "text": "JMS poison message sentinel could not audit destination '{{ inputs.dlq_destination_name }}' on broker {{ inputs.jms_broker_url }}. Check broker connectivity, credentials, and provider_jar_path. Execution {{ execution.id }}."
            }
    defaults:
      - id: log_unclassified
        type: io.kestra.plugin.core.log.Log
        description: "Fallback branch; reached only if classify_quarantine_status
          produces a value outside the three known cases."
        message: "JMS poison message sentinel produced an unclassified status for
          destination {{ inputs.dlq_destination_name }}: {{
          outputs.classify_quarantine_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: "JMS poison message audit completed for destination '{{
      inputs.dlq_destination_name }}'. Status: {{
      outputs.classify_quarantine_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: JMS poison message quarantine gate failed during audit of destination '{{ inputs.dlq_destination_name }}' on broker {{ inputs.jms_broker_url }}. Execution ID: {{ execution.id }}."
      }

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

A dead-letter queue filling up with ordinary failures is expected; a dead-letter queue filling up with the same poison messages over and over is a different, more urgent problem; the two look identical unless you check how many times each message has already been redelivered. This blueprint uses plugin-jms's native Consume task as the probe, filtering server-side with a JMS message selector on the standard JMSXDeliveryCount property so only genuinely poisoned messages, those redelivered past a threshold, count toward the alert.

How it works

audit_poison_messages (io.kestra.plugin.jms.Consume) connects to dlq_destination_name via a DIRECT connectionFactoryConfig, applies the message selector JMSXDeliveryCount > poison_redelivery_threshold, and pulls up to max_messages_per_audit matches within max_wait_timeout_ms. destinationType is hard-coded to QUEUE: it is a fixed Java enum on the task, not a Property<String>, so Kestra cannot render a Pebble expression into it, and dead-letter destinations are a point-to-point consume pattern in practice, never a pub-sub TOPIC. allowFailure: true lets the flow continue instead of aborting if the broker is unreachable or the provider JAR cannot be loaded.
classify_quarantine_status (io.kestra.plugin.core.debug.Return) checks whether outputs.audit_poison_messages.count is defined at all. If it is missing, the classification is PROBE_FAILED. If it is defined and at or above poison_alert_threshold, the classification is POISON_QUARANTINE_BREACH; otherwise HEALTHY.
route_by_quarantine_status (io.kestra.plugin.core.flow.Switch) branches on that string. POISON_QUARANTINE_BREACH runs evaluate_alert_toggle (io.kestra.plugin.core.flow.If), which pages Slack via dispatch_poison_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 quarantine 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 poison-message trend.
The flow-level errors block pages Slack separately if the sentinel flow itself fails outright.
The jms_poison_audit_schedule trigger polls every 6 hours with explicit input overrides, shipped disabled: true until pointed at a real broker and destination.

What you get

A scheduled, selector-filtered poison-message audit against any JMS-compliant broker (ActiveMQ, Artemis, and other providers with a matching ConnectionFactory class and client JAR), with zero custom consumer code.
A three-way status output (POISON_QUARANTINE_BREACH / HEALTHY / PROBE_FAILED) other flows can gate on.
Server-side filtering by JMSXDeliveryCount so the alert count reflects genuinely stuck messages, not every routine dead-letter arrival.
A Slack page naming the destination, matched poison count, threshold, and broker on breach or probe failure.
An internal-storage URI (outputs.audit_poison_messages.uri) containing the matched poison 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 ActiveMQ, Artemis, IBM MQ, or other JMS-compliant brokers in production.
Integration and middleware teams who need to distinguish a one-off dead-letter arrival from a message that keeps failing redelivery after redelivery.
Platform teams standardizing Day 2 backlog and compliance governance across message-bus-backed services inside Kestra.

Why orchestrate this with Kestra

A one-off JMX console check or a custom cron script proves the dead-letter count for one instant and loses the redelivery-count context. This flow keeps every audit in Kestra's execution history, turns the result into a typed status output other flows can depend on, stores the matched poison 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 JMS-compliant broker reachable from the Kestra Worker, with a dead-letter destination already receiving redelivered messages (ActiveMQ Classic and Artemis both populate the standard JMSXDeliveryCount property on redelivery).
The JMS provider's client JAR made available either under the plugin's jms-libs folder on the Worker, or uploaded as a Kestra namespace file and referenced via provider_jar_path.
Broker credentials with consume permission on the dead-letter destination.
A Slack incoming webhook for the breach, probe-failure, and flow-failure alerts.

Quick start

Add the JMS_USERNAME, JMS_PASSWORD, and SLACK_WEBHOOK secrets.
Point dlq_destination_name, jms_broker_url, and jms_connection_factory_class at your real broker and dead-letter destination.
Run the flow once and confirm status reads HEALTHY or POISON_QUARANTINE_BREACH as expected for your current redelivery counts.
Enable jms_poison_audit_schedule.

How to extend

Parse the matched batch at outputs.audit_poison_messages.uri to group poison messages by headers (JMS properties) such as originating queue or error type.
Push status into a KV store to alert only on state transitions instead of every run.
Fan the probe out to several dead-letter destinations with a Loop task to monitor multiple services from one flow.
Chain a io.kestra.plugin.jms.Produce task to republish selectively reprocessed messages back onto their source destination after manual review.

Secrets

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

Inputs

jms_broker_url (STRING, default tcp://localhost:61616): broker connection URL.
jms_connection_factory_class (STRING, default org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory): ConnectionFactory implementation class.
provider_jar_path (STRING, default kestra:///jms/artemis-client.jar): provider client JAR location.
dlq_destination_name (STRING, default ActiveMQ.DLQ): dead-letter destination to audit. Always treated as a QUEUE, see Pitfalls.
message_serde_type (SELECT, one of STRING, JSON, BYTES, default STRING): message body deserializer.
poison_redelivery_threshold (INT, default 3): JMSXDeliveryCount floor used in the selector to define poison.
max_messages_per_audit (INT, default 50): maximum matching messages pulled per audit pass.
max_wait_timeout_ms (INT, default 5000): bounded poll time in milliseconds.
poison_alert_threshold (INT, default 5): matched count at or above which the destination is classified POISON_QUARANTINE_BREACH.
send_slack_alert (STRING, default "true"): when "true", page Slack on breach; when "false", only log it.

Outputs

{{ outputs.audit_poison_messages.count }}: number of matched poison messages in this pass.
{{ outputs.audit_poison_messages.uri }}: internal storage URI (kestra://) containing the matched message batch.
{{ outputs.classify_quarantine_status.value }}: classification string, one of POISON_QUARANTINE_BREACH, HEALTHY, PROBE_FAILED.
{{ outputs.status }}: flow-level alias of the same classification.

Pitfalls

destinationType on io.kestra.plugin.jms.Consume is a fixed enum (QUEUE or TOPIC), not a Property<String>, so Kestra cannot render a Pebble input into it; a templated {{ inputs.destination_type }} fails deserialization before the flow ever runs. This blueprint hard-codes QUEUE. If you need to audit a TOPIC-backed dead-letter destination, branch with an io.kestra.plugin.core.flow.If into two separate Consume tasks, one with destinationType: QUEUE and one with destinationType: TOPIC, each hard-coded.
maxWaitTimeout defaults to 0 inside plugin-jms, which blocks indefinitely waiting for a match. This blueprint always sets max_wait_timeout_ms explicitly; do not clear it to 0 in a scheduled flow.
Consuming from the dead-letter destination acknowledges and permanently removes the matched messages (CLIENT_ACKNOWLEDGE, at-least-once); this is not a non-destructive peek. Only point this flow at a dedicated dead-letter or quarantine destination.
JMSXDeliveryCount is a standard optional JMS property; it is populated by ActiveMQ Classic and Artemis on redelivery, but a non-compliant or misconfigured provider may not set it, which would make every message match or none at all. Verify the property is present before trusting the threshold.
The plugin's own inline Javadoc example shows the destination field as name; the real field verified against the plugin's Java source and test suite is destinationName. Use destinationName, not name.
The JMS provider's client JAR must be resolvable at provider_jar_path or under the plugin's jms-libs folder; a missing or mismatched JAR surfaces as a connection failure, which this flow classifies as PROBE_FAILED rather than HEALTHY.
connectionFactoryClass must match the broker you actually run; the Artemis and ActiveMQ Classic connection factory classes are not interchangeable.

Local Container Setup

To start a local Apache ActiveMQ Artemis broker container for testing, execute:

docker run --detach --name kestra-artemis -p 61616:61616 -p 8161:8161 apache/activemq-artemis:latest-alpine

The default web console credentials are username and password both set to artemis; override them with -e ARTEMIS_USER= -e ARTEMIS_PASSWORD= if needed.

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

docker network connect YOUR_KESTRA_NETWORK kestra-artemis

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

JMS Poison Message Quarantine 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?

JMS Poison Message Quarantine 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

Meilisearch Indexing Queue Staleness Sentinel

MinIO Cluster Disk Healing Sentinel

Prune noncurrent object versions billed forever behind a versioned bucket

Enforce default SSE-S3 encryption on a MinIO bucket

Catch MinIO Replication Drift Before Objects Go Missing on the Target

Meilisearch Index Health Sentinel and Search Audit

More Data Blueprints

Huawei DLI Data Quality Assertion Gate

InfluxDB Bucket Retention Sentinel

Idempotent API Poller with Dynamic KV Watermark

Load time series data into InfluxDB

Huawei GeminiDB TTL Hygiene Sentinel

Sync Hubspot CRM data to BigQuery on a schedule

New to Kestra?