Fan Out Pipeline Completion Events to NATS Subscribers

Stop making every downstream team poll the warehouse or query Kestra to find out a nightly pipeline finished. This blueprint refreshes a daily sales aggregate in Postgres on a schedule, then publishes three purpose-built completion events onto three independent NATS subjects, one for fulfillment, one for analytics, and one for audit, each carrying only the fields that team actually needs. Every subscriber reacts to the same pipeline run without coupling to the database schema or to each other.

How it works

1. refresh_daily_sales (io.kestra.plugin.jdbc.postgresql.Query) upserts today's row into reporting.daily_sales, aggregating order count and revenue from public.orders for the current date.
2. read_aggregate (io.kestra.plugin.jdbc.postgresql.Query) reads that row back with fetchType: FETCH_ONE, exposing total_orders and total_revenue as outputs.read_aggregate.row.*.
3. notify_fulfillment (io.kestra.plugin.nats.core.Produce) publishes a small JSON payload with just the order count onto pipeline.completed.fulfillment.
4. notify_analytics (io.kestra.plugin.nats.core.Produce) publishes a payload with just the revenue total onto pipeline.completed.analytics.
5. notify_audit (io.kestra.plugin.nats.core.Produce) publishes the full aggregate plus the Kestra execution id onto pipeline.completed.audit, giving compliance a complete, traceable record.
6. The nightly Schedule trigger runs the whole sequence once a day; the errors block posts a Slack alert if the refresh or any publish step fails, so a broken fan-out is never silent.

Postgres and NATS connection details are each set once through pluginDefaults, so the three Produce tasks only need to declare their subject and payload.

What you get

• A single pipeline run that reliably notifies three independent downstream teams without any of them polling the database or Kestra's API.
• Purpose-shaped payloads per subject, so each subscriber gets exactly the fields it needs instead of parsing a generic completion event.
• A durable audit trail on pipeline.completed.audit that ties every completion event back to a specific Kestra execution id.
• A pattern that scales to a fourth or fifth downstream team by adding one more Produce task, with zero changes to the existing subscribers.

Who it's for

• Data platform teams running NATS as their internal event bus who need to decouple a scheduled ETL job from its downstream consumers.
• Analytics and fulfillment engineering teams who currently poll a warehouse table to detect that a nightly job finished.
• Compliance and audit teams who need a durable, subject-based record of every pipeline completion tied to an execution id.

Why orchestrate this with Kestra

NATS delivers messages reliably once you publish them, but it has no scheduler, no SQL execution, and no concept of "refresh this aggregate, then tell three different subjects about it, and alert someone if any step fails." Kestra supplies all of that: the Schedule trigger drives the refresh, the JDBC tasks compute the aggregate the fan-out actually reports on, three Produce tasks publish tailored events as one auditable execution, and the errors block guarantees a failure anywhere in the chain still reaches a human. The whole notification contract for three teams lives in one version-controlled YAML file instead of being scattered across cron jobs and ad hoc publish scripts.

Prerequisites

• A PostgreSQL database with a public.orders table and a reporting.daily_sales table keyed on report_date.
• A running NATS server reachable at the configured url.
• Downstream services subscribed to pipeline.completed.fulfillment, pipeline.completed.analytics, and pipeline.completed.audit.
• A Slack incoming webhook for failure alerts.

Secrets

• POSTGRES_URL: JDBC URL, for example jdbc:postgresql://warehouse-host:5432/orders.
• POSTGRES_USERNAME / POSTGRES_PASSWORD: database credentials.
• NATS_URL: NATS server URL, for example nats://nats.internal:4222.
• NATS_USERNAME / NATS_PASSWORD: credentials for the NATS server.
• SLACK_WEBHOOK_URL: Slack incoming webhook URL.

Quick start

1. Add the six secrets above to your Kestra namespace.
2. Confirm the public.orders and reporting.daily_sales tables exist with the expected columns.
3. Subscribe at least one test consumer to each of the three subjects (for example nats sub "pipeline.completed.>").
4. Set disabled: false on the nightly trigger, or run the flow manually to see all three events published.

How to extend

• Add a fourth Produce task for a new downstream team without touching the existing three, since each subscriber only depends on its own subject.
• Replace the fixed pipeline.completed.* subjects with a templated subject per region or business unit for multi-tenant fan-out.
• Swap the Postgres aggregate for a dbt run or a warehouse query, keeping the fan-out step unchanged.
• Add a NATS Key/Value write (io.kestra.plugin.nats.kv.Put) alongside the fan-out so late subscribers can read the last completion state instead of only reacting to the live event.

Links

• Plugin docs: https://kestra.io/plugins/plugin-nats