NATS Event Validation and Quarantine with DuckDB in Kestra

id: nats-event-validation-quarantine
namespace: company.team

description: |
  Validate incoming NATS JetStream sensor events using DuckDB in-memory SQL
  processing and route non-compliant records to a quarantine topic with Slack alerts.

inputs:
  - id: nats_url
    type: STRING
    defaults: nats://nats-js:4222
    description: NATS server connection URL.

  - id: seed_demo_events
    type: BOOLEAN
    defaults: true
    description: Whether to publish seed demo events before running validation.

  - id: max_events
    type: SELECT
    defaults: "100"
    values:
      - "50"
      - "100"
      - "500"
    description: Maximum number of records to pull from NATS JetStream per batch.

  - id: min_temperature
    type: SELECT
    defaults: "-50"
    values:
      - "-50"
      - "-20"
      - "0"
    description: Minimum allowable sensor temperature in degrees Celsius.

  - id: max_temperature
    type: SELECT
    defaults: "200"
    values:
      - "100"
      - "150"
      - "200"
    description: Maximum allowable sensor temperature in degrees Celsius.

triggers:
  - id: hourly_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    description: Periodically check NATS JetStream for incoming sensor events to validate.
    cron: "0 * * * *"
    disabled: true
    inputs:
      seed_demo_events: false

tasks:
  - id: seed_events
    type: io.kestra.plugin.core.flow.If
    description: Conditionally publish seed demo events to NATS JetStream for
      validation testing.
    condition: "{{ inputs.seed_demo_events }}"
    then:
      - id: produce_seed_events
        type: io.kestra.plugin.nats.core.Produce
        description: Publish sample IoT sensor events to NATS JetStream subject
          events.sensor.
        url: "{{ inputs.nats_url }}"
        subject: events.sensor
        from:
          - data: '{"event_id":"evt-001","sensor_id":"S1","temperature":22.5,"unit":"celsius"}'
          - data: '{"event_id":"evt-002","sensor_id":"S5","temperature":-5.0,"unit":"celsius"}'
          - data: '{"event_id":"evt-003","sensor_id":"S2","temperature":null,"unit":"celsius"}'
          - data: '{"event_id":"evt-004","sensor_id":"S3","temperature":999.0,"unit":"celsius"}'
          - data: '{"event_id":"evt-005","sensor_id":"S4","temperature":18.0,"unit":"fahrenheit"}'
          - data: '{"event_id":"","sensor_id":"S6","temperature":21.0,"unit":"celsius"}'
          - data: 'MALFORMED_JSON_PAYLOAD_TEST_STRING'

  - id: consume_events
    type: io.kestra.plugin.nats.core.Consume
    description: Pull incoming sensor events from NATS JetStream subject events.sensor.
    url: "{{ inputs.nats_url }}"
    subject: events.sensor
    durableId: kestra-event-validator
    maxRecords: "{{ inputs.max_events | int }}"
    maxDuration: PT5S
    batchSize: 100

  - id: check_messages
    type: io.kestra.plugin.core.flow.If
    description: Process consumed events if messages arrive, or log a notice when
      zero events are pulled.
    condition: "{{ outputs.consume_events.messagesCount > 0 }}"
    then:
      - id: convert_ion_to_json
        type: io.kestra.plugin.serdes.json.IonToJson
        description: Convert internal Ion format from NATS consume task to
          newline-delimited JSON.
        from: "{{ outputs.consume_events.uri }}"

      - id: valid_events
        type: io.kestra.plugin.jdbc.duckdb.Query
        description: Filter valid events matching schema rules and temperature
          thresholds using DuckDB.
        inputFiles:
          events.json: "{{ outputs.convert_ion_to_json.uri }}"
        fetchType: STORE
        sql: |
          WITH classified AS (
              SELECT
                  data,
                  CASE
                      WHEN NOT json_valid(data) THEN 'MALFORMED_JSON'
                      WHEN json_extract_string(data, '$.event_id') IS NULL OR trim(json_extract_string(data, '$.event_id')) = '' THEN 'MISSING_EVENT_ID'
                      WHEN json_extract_string(data, '$.sensor_id') IS NULL OR trim(json_extract_string(data, '$.sensor_id')) = '' THEN 'MISSING_SENSOR_ID'
                      WHEN json_extract_string(data, '$.temperature') IS NULL OR TRY_CAST(json_extract_string(data, '$.temperature') AS DOUBLE) IS NULL THEN 'NULL_TEMPERATURE'
                      WHEN TRY_CAST(json_extract_string(data, '$.temperature') AS DOUBLE) < CAST('{{ inputs.min_temperature }}' AS DOUBLE) OR TRY_CAST(json_extract_string(data, '$.temperature') AS DOUBLE) > CAST('{{ inputs.max_temperature }}' AS DOUBLE) THEN 'TEMPERATURE_OUT_OF_RANGE'
                      WHEN json_extract_string(data, '$.unit') IS NULL OR json_extract_string(data, '$.unit') != 'celsius' THEN 'INVALID_UNIT'
                      ELSE NULL
                  END AS reason
              FROM read_json_auto('{{ workingDir }}/events.json', columns={subject: 'VARCHAR', data: 'VARCHAR'})
          )
          SELECT data FROM classified WHERE reason IS NULL;

      - id: quarantined_events
        type: io.kestra.plugin.jdbc.duckdb.Query
        description: Extract invalid events and append validation failure reasons using
          DuckDB.
        inputFiles:
          events.json: "{{ outputs.convert_ion_to_json.uri }}"
        fetchType: STORE
        sql: |
          WITH classified AS (
              SELECT
                  data,
                  CASE
                      WHEN NOT json_valid(data) THEN 'MALFORMED_JSON'
                      WHEN json_extract_string(data, '$.event_id') IS NULL OR trim(json_extract_string(data, '$.event_id')) = '' THEN 'MISSING_EVENT_ID'
                      WHEN json_extract_string(data, '$.sensor_id') IS NULL OR trim(json_extract_string(data, '$.sensor_id')) = '' THEN 'MISSING_SENSOR_ID'
                      WHEN json_extract_string(data, '$.temperature') IS NULL OR TRY_CAST(json_extract_string(data, '$.temperature') AS DOUBLE) IS NULL THEN 'NULL_TEMPERATURE'
                      WHEN TRY_CAST(json_extract_string(data, '$.temperature') AS DOUBLE) < CAST('{{ inputs.min_temperature }}' AS DOUBLE) OR TRY_CAST(json_extract_string(data, '$.temperature') AS DOUBLE) > CAST('{{ inputs.max_temperature }}' AS DOUBLE) THEN 'TEMPERATURE_OUT_OF_RANGE'
                      WHEN json_extract_string(data, '$.unit') IS NULL OR json_extract_string(data, '$.unit') != 'celsius' THEN 'INVALID_UNIT'
                      ELSE NULL
                  END AS reason
              FROM read_json_auto('{{ workingDir }}/events.json', columns={subject: 'VARCHAR', data: 'VARCHAR'})
          )
          SELECT json_object('reason', reason, 'original', data)::VARCHAR AS data FROM classified WHERE reason IS NOT NULL;

      - id: publish_valid_events
        type: io.kestra.plugin.core.flow.If
        description: Publish validated sensor events to events.processed when valid
          records exist.
        condition: "{{ outputs.valid_events.size > 0 }}"
        then:
          - id: produce_valid_events
            type: io.kestra.plugin.nats.core.Produce
            description: Publish validated sensor events to NATS subject events.processed.
            url: "{{ inputs.nats_url }}"
            subject: events.processed
            from: "{{ outputs.valid_events.uri }}"

      - id: handle_quarantine_events
        type: io.kestra.plugin.core.flow.If
        description: Publish quarantined events and send Slack notification when invalid
          records exist.
        condition: "{{ outputs.quarantined_events.size > 0 }}"
        then:
          - id: produce_quarantine_events
            type: io.kestra.plugin.nats.core.Produce
            description: Publish quarantined events with error details to NATS subject
              events.quarantine.
            url: "{{ inputs.nats_url }}"
            subject: events.quarantine
            from: "{{ outputs.quarantined_events.uri }}"

          - id: notify_quarantine_slack
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            description: Post quarantine notification summary to Slack webhook channel.
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            payload: |
              {"text": "NATS Event Validation Alert: {{ outputs.quarantined_events.size }} of {{ outputs.consume_events.messagesCount }} events quarantined on subject events.sensor."}

      - id: log_summary
        type: io.kestra.plugin.core.log.Log
        description: Output processing statistics for consumed, valid, and quarantined
          event counts.
        message: "Event validation completed. Total consumed: {{
          outputs.consume_events.messagesCount }}, Valid: {{
          outputs.valid_events.size }}, Quarantined: {{
          outputs.quarantined_events.size }}."
    else:
      - id: log_no_events
        type: io.kestra.plugin.core.log.Log
        description: Log message when no messages are fetched from NATS.
        message: "No new events consumed from NATS subject events.sensor."

errors:
  - id: notify_failure_slack
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Send critical failure alert to Slack if flow execution errors.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {"text": "Flow execution failed for nats-event-validation-quarantine (Execution ID: {{ execution.id }})."}

Diagram unavailable

We could not build the topology for this blueprint. The flow itself is valid, use the YAML on the left to run it.

Validate incoming NATS JetStream sensor events using DuckDB in-memory SQL processing and route non-compliant records to a quarantine topic with Slack alerts.

Who it's for

Data Engineers and Event-Driven Architecture Operators who require automated quality gates, schema enforcement, and payload validation for real-time messaging streams.
Infrastructure teams managing NATS JetStream messaging clusters who need inline stream cleansing before writing records to downstream databases.
System Reliability Engineers who need real-time Slack notifications when telemetry streams contain payload anomalies or malformed JSON messages.

Why orchestrate this with Kestra

NATS JetStream ingests streaming data at high velocity, but it has no native schema validation or SQL query engine. Hardcoding schema validation inside downstream microservices leads to duplicated code, fragile exception handling, and silent data corruption. Kestra solves this by orchestrating the entire lifecycle: consuming batches from JetStream via a durable consumer, running zero-setup DuckDB SQL quality checks in memory, routing clean records to events.processed, writing invalid payloads to events.quarantine, and alerting engineers over Slack.

How it works

seed_events (io.kestra.plugin.core.flow.If): When seed_demo_events input is true, publishes 7 sample sensor events (2 valid, 5 invalid/malformed) to events.sensor using io.kestra.plugin.nats.core.Produce.
consume_events (io.kestra.plugin.nats.core.Consume): Pulls a batch of messages from events.sensor using durable consumer kestra-event-validator and writes raw records to Kestra internal storage in Ion format.
check_messages (io.kestra.plugin.core.flow.If): Evaluates outputs.consume_events.messagesCount > 0. If false, logs a notice and skips processing.
convert_ion_to_json (io.kestra.plugin.serdes.json.IonToJson): Converts internal Ion storage records into newline-delimited JSON for DuckDB query consumption.
valid_events (io.kestra.plugin.jdbc.duckdb.Query): Executes DuckDB SQL classification query to extract rows where reason IS NULL (valid event_id, sensor_id, temperature within range, unit celsius).
quarantined_events (io.kestra.plugin.jdbc.duckdb.Query): Executes DuckDB SQL classification query to extract rows where reason IS NOT NULL (MALFORMED_JSON, MISSING_EVENT_ID, MISSING_SENSOR_ID, NULL_TEMPERATURE, TEMPERATURE_OUT_OF_RANGE, INVALID_UNIT), outputting JSON objects containing the error reason and original payload.
publish_valid_events (io.kestra.plugin.core.flow.If): When valid_events.size > 0, publishes validated records to NATS subject events.processed.
handle_quarantine_events (io.kestra.plugin.core.flow.If): When quarantined_events.size > 0, publishes quarantined records to NATS subject events.quarantine and sends a summary notification via io.kestra.plugin.slack.notifications.SlackIncomingWebhook.
log_summary (io.kestra.plugin.core.log.Log): Logs total consumed, valid, and quarantined event counts.
errors: Contains a SlackIncomingWebhook task to alert SREs if any workflow task fails unexpectedly.

Prerequisites

A running NATS JetStream server (e.g., container named nats-js).
A NATS Stream configured to capture subjects matching events.>.
Connectivity between the Kestra worker container and the NATS server network.
A Slack Incoming Webhook URL configured in Kestra secrets.

Setup Commands

Start a NATS JetStream container on your Kestra network:

docker run -d --name nats-js --network <kestra_network> -p 4222:4222 -p 8222:8222 nats:latest -js

Create JetStream stream for events.> using nats-box CLI container:

docker run --rm --network <kestra_network> natsio/nats-box nats -s nats://nats-js:4222 stream add SENSOR_EVENTS --subjects "events.>" --storage file --defaults

Secrets

SLACK_WEBHOOK_URL: Slack Incoming Webhook URL. (Note for open-source users: Store secrets in environment variables prefixed with SECRET_ encoded in base64, e.g. SECRET_SLACK_WEBHOOK_URL)

Quick start

Configure nats_url input (default: nats://nats-js:4222).
Keep seed_demo_events: true for initial validation run.
Execute the flow and inspect valid records in events.processed, quarantined events in events.quarantine, and Slack channel notifications.

How to extend

Add custom DuckDB classification branches for additional sensor parameters like pressure, humidity, or battery percentage.
Add a Postgres or Snowflake sink task (io.kestra.plugin.jdbc.postgresql.Query) after publish_valid_events to persist clean records directly into a data warehouse.
Replace Slack webhook notifications with PagerDuty alerts or Microsoft Teams webhooks using Kestra plugin tasks.
Configure automated dead-letter queue replay by building a secondary flow that reads from events.quarantine, fixes payload schemas, and re-publishes to events.sensor.

Pitfalls

At-Least-Once Acknowledgment: Messages are acknowledged upon consume fetch. If a downstream task fails, the consumed batch is not redelivered automatically.
Core NATS Publishing: The Produce task publishes using core NATS. If no JetStream stream captures events.processed or events.quarantine, published messages are silently dropped.
Stream Requirement: The target stream must exist in NATS before executing the Consume task.
Exact Subject Filter: The consumer filters specifically on events.sensor to prevent infinite feedback loops with events.processed.
Malformed Payload Handling: Non-JSON payloads land in quarantine with MALFORMED_JSON without failing the flow execution.

What you get

A single, fully automated event quality pipeline that consumes, validates, and routes NATS JetStream messages without custom microservice code.
Zero-copy SQL validation powered by DuckDB in-memory engine, performing schema enforcement directly on streaming payloads.
Complementary partitioning of event batches into valid payloads and quarantined anomaly records with detailed root cause tags.
Immediate Slack alerting for quarantine spikes and execution failure tracking across your message bus.