Google Cloud RealtimeTrigger

Google Cloud RealtimeTrigger
Certified
Realtime Pub/Sub trigger

Subscribes to a Pub/Sub subscription and creates one execution per received message. Use the batch Trigger for grouped consumption.

yaml
type: io.kestra.plugin.gcp.pubsub.RealtimeTrigger

Examples

Consume a message from a Pub/Sub topic in real-time

yaml
id: realtime_pubsub
namespace: company.team

tasks:
  - id: log
    type: io.kestra.plugin.core.log.Log
    message: "Received: {{ trigger.data }}"

triggers:
  - id: trigger
    type: io.kestra.plugin.gcp.pubsub.RealtimeTrigger
    projectId: test-project-id
    topic: test-topic
    subscription: test-subscription

Use GCP Pub/Sub Realtime Trigger to push events into Firestore

yaml
id: pubsub_realtime_trigger
namespace: company.team

tasks:
  - id: insert_into_firestore
    type: io.kestra.plugin.gcp.firestore.Set
    projectId: test-project-id
    collection: orders
    document:
      order_id: "{{ trigger.data | jq('.order_id') | first }}"
      customer_name: "{{ trigger.data | jq('.customer_name') | first }}"
      customer_email: "{{ trigger.data | jq('.customer_email') | first }}"
      product_id: "{{ trigger.data | jq('.product_id') | first }}"
      price: "{{ trigger.data | jq('.price') | first }}"
      quantity: "{{ trigger.data | jq('.quantity') | first }}"
      total: "{{ trigger.data | jq('.total') | first }}"

triggers:
  - id: realtime_trigger
    type: io.kestra.plugin.gcp.pubsub.RealtimeTrigger
    projectId: test-project-id
    topic: orders
    subscription: kestra-subscription
    serdeType: JSON

Properties

topic *string

The Pub/Sub topic

The Pub/Sub topic. It must be created before executing the task.

allowConcurrent boolean

Defaultfalse

Specifies whether a trigger is allowed to start a new execution even if a previous run is still in progress.

autoCreateSubscription booleanstring

Defaulttrue

Auto-create subscription

Create the subscription if missing; default true

impersonatedServiceAccount string

The GCP service account to impersonate

interval string

DefaultPT1M

How often the trigger checks for new messages

maxDuration string

Max duration

Optional run duration limit (ISO-8601)

maxRecords integerstring

Max records

Optional cap on messages before stopping

projectId string

The GCP project ID

scopes array

SubTypestring

Default["https://www.googleapis.com/auth/cloud-platform"]

The GCP scopes to be used

serdeType string

DefaultSTRING

Possible Values

STRINGJSON

Serde type

Serializer/deserializer for messages; defaults to STRING

serviceAccount string

The GCP service account

stopAfter array

SubTypestring

Possible Values

CREATEDSUBMITTEDRUNNINGPAUSEDRESTARTEDKILLINGSUCCESSWARNINGFAILEDKILLEDCANCELLEDQUEUEDRETRYINGRETRIEDSKIPPEDBREAKPOINTRESUBMITTED

List of execution states after which a trigger should be stopped (a.k.a. disabled).

subscription string

Subscription

Subscription name; auto-created when autoCreateSubscription is true

when string

Defaulttrue

A condition that determines whether the trigger should run.

A Pebble expression evaluated at trigger time. The trigger fires only when the expression evaluates to a truthy value (true, a non-empty string, a non-zero number). Use this to gate trigger execution on dynamic runtime values such as execution labels, flow variables, or environment conditions.

Outputs

attributes object

The message attributes map

data object

The message data, must be a string if serde type is 'STRING', otherwise a JSON object

If it's a string, it can be a dynamic property otherwise not.

messageId string

The message identifier

orderingKey string

The message ordering key