Process Pub/Sub Messages into Postgres in Real Time

Stream Google Cloud Pub/Sub messages straight into Postgres in real time, turning every message into a governed, observable Kestra execution. This event-driven blueprint solves the classic gap between a high-throughput message bus and a relational database: instead of writing and babysitting a custom consumer service, you get one execution per message, payload validation, safe parameterized upserts, full run history, and automatic failure alerts to Microsoft Teams. It is ideal for ingesting order events, IoT readings, clickstream data, or any JSON message stream that needs to land in Postgres with auditability.

How it works

1. The io.kestra.plugin.gcp.pubsub.RealtimeTrigger listens on the orders topic via the kestra-orders-sub subscription and starts one execution per message. With serdeType: JSON, each payload is parsed so its fields are directly addressable as trigger.data.*.
2. The route task (io.kestra.plugin.core.flow.If) checks trigger.data.event_type == 'order' and only lets order events proceed, skipping everything else.
3. The upsert_order task (io.kestra.plugin.jdbc.postgresql.Query) runs an INSERT ... ON CONFLICT (order_id) DO UPDATE against Postgres, binding order_id, customer, and amount through named parameters so values are never concatenated into SQL.
4. If a message fails to land, the alert_on_failure error handler (io.kestra.plugin.microsoft365.teams.TeamsIncomingWebhook) posts a notification to Microsoft Teams with the failing execution id.

What you get

• One isolated execution per Pub/Sub message, each with its own logs and history.
• Idempotent upserts that safely handle duplicate or replayed messages.
• SQL injection protection through named parameter binding.
• Built-in failure alerting to Microsoft Teams.
• A clear, auditable record of every message that did (or did not) reach the database.

Who it's for

• Data engineers building real-time ingestion from Pub/Sub into a relational store.
• Platform teams replacing bespoke consumer microservices with declarative, observable flows.
• Backend teams that need order, event, or telemetry data landed reliably in Postgres.

Why orchestrate this with Kestra

Pub/Sub delivers messages, but it has no scheduler, no per-message run history, no validation step, and no built-in retry or alerting around the database write. Kestra fills that gap: the event trigger reacts the moment a message arrives, each message becomes a first-class execution you can inspect and replay, step-level retries and the error handler protect the load, and lineage across the trigger, validation, and upsert is captured automatically. It is all declared in readable YAML, so the pipeline is versioned and reviewable rather than buried in consumer code.

Prerequisites

A Pub/Sub topic (orders) with a subscription, and a Postgres orders table with an order_id primary key.

Secrets

• GCP_PROJECT_ID: GCP project hosting the Pub/Sub topic.
• GCP_SERVICE_ACCOUNT: service account key JSON with Pub/Sub subscribe access.
• POSTGRES_URL: JDBC URL, e.g. jdbc:postgresql://host:5432/db.
• POSTGRES_USERNAME: database username.
• POSTGRES_PASSWORD: database password.
• TEAMS_WEBHOOK_URL: Microsoft Teams incoming webhook URL.

Quick start

1. Set the secrets above in your Kestra namespace.
2. Point the trigger topic and subscription at your Pub/Sub resources.
3. Adjust the route condition, the SQL, and the parameters to match your schema.
4. Enable the trigger (it ships disabled: true so importing the blueprint does not connect with placeholder config).

How to extend

• Change the route condition to fan out by event type, or replace If with a Switch to route different events to different tables.
• Swap the upsert SQL for a bulk batch insert, or add a downstream transformation task before the write.
• Add more error handlers (Slack, email, PagerDuty) alongside the Teams alert.
• Reuse the same pattern to push into BigQuery, GCS, or any other JDBC database.

Links

• Plugin docs: plugin-gcp
• Pub/Sub orchestration with Kestra