Skip to main content

Google Cloud Pub/Sub

Synopsis​

Creates a subscriber that connects to a Google Cloud Pub/Sub subscription and consumes messages using concurrent goroutine receivers. Supports Application Default Credentials and explicit service account credentials.

Schema​

- id: <numeric>
  name: <string>
  description: <string>
  type: pubsub
  tags: <string[]>
  pipelines: <pipeline[]>
  status: <boolean>
  properties:
    project_id: <string>
    subscription_id: <string>
    credentials_json: <string>
    workers: <numeric>
    max_outstanding_messages: <numeric>
    max_outstanding_bytes: <numeric>

Configuration​

The following fields are used to define the device:

Device​

FieldRequiredDefaultDescription
idY-Unique numeric identifier
nameY-Device name
descriptionN-Optional description
typeY-Must be pubsub
tagsN-Optional tags
pipelinesN-Optional pre-processor pipelines
statusNtrueEnable/disable the device

Connection​

FieldRequiredDefaultDescription
project_idYGoogle Cloud project ID
subscription_idYPub/Sub subscription name
credentials_jsonN-Service account credentials JSON. When omitted, Application Default Credentials are used

Performance​

FieldRequiredDefaultDescription
workersN0Number of goroutines for concurrent message processing. 0 defaults to GOMAXPROCS
max_outstanding_messagesN1000Maximum number of unprocessed messages held in memory at once
max_outstanding_bytesN1000000000Maximum total byte size of unprocessed messages held in memory at once (1 GB)
max_buffer_sizeN128MBIngest buffer admission cap. Also sizes this listener's ingest buffers: 3/8 of this value at start, growing to at most 3x
note

max_buffer_size is the ingest-buffer admission cap, and is distinct from any buffer_size field above, which sizes the network read buffer. Left unset it is 128MB, whose buffers take 48 MiB of heap per listener worker at start and grow to at most 384 MiB as bursts need the room — and with reuse: true a device runs one worker per listener. Size strings are binary, and 32MB and 32MiB are exact synonyms.

See Listener Memory Sizing for the arithmetic, the sizing table, and what happens when the cap is exhausted.

Details​

The device connects to a single Pub/Sub subscription and opens a long-lived streaming pull connection. Message delivery uses the Google Cloud Pub/Sub SDK's Receive method, which dispatches the callback concurrently from multiple goroutines controlled by the workers setting. A mutex serializes all writes to the internal stream writers, so concurrent callbacks do not overwrite each other's data.

Messages are acknowledged after a successful write. If the write fails, the message is negatively acknowledged and redelivered. Empty payloads are acknowledged immediately without processing to prevent infinite redelivery loops.

The workers field controls the number of concurrent goroutine receivers within a single Receive call. This is the correct scaling mechanism for Pub/Sub throughput: multiple Receive calls on the same subscription compete for the same messages rather than parallelizing work, whereas increasing workers lets a single subscriber drain faster.

The max_outstanding_messages and max_outstanding_bytes limits cap how many messages are held in memory concurrently, bounding memory consumption under burst conditions. Setting these values too high can cause out-of-memory errors during traffic spikes.

When credentials_json is omitted, the SDK resolves credentials using Application Default Credentials, which includes Workload Identity Federation on GKE and other Google-managed environments. When running outside Google Cloud, a service account key JSON must be supplied explicitly.

The subscription must already exist in the specified project. The device validates both project_id and subscription_id at startup and will not start if either is missing or empty. The connection state is set to connected once the consumer is successfully created.

Examples​

Basic​

Creating a basic Pub/Sub subscriber using Application Default Credentials...

- id: 1
  name: pubsub_basic
  type: pubsub
  properties:
    project_id: "my-gcp-project"
    subscription_id: "my-subscription"

Service Account​

Connecting with an explicit service account key...

- id: 2
  name: pubsub_service_account
  type: pubsub
  properties:
    project_id: "my-gcp-project"
    subscription_id: "my-subscription"
    credentials_json: |
      {
        "type": "service_account",
        "project_id": "my-gcp-project",
        "private_key_id": "key-id",
        "private_key": "-----BEGIN RSA PRIVATE KEY-----\n...\n-----END RSA PRIVATE KEY-----\n",
        "client_email": "datastream@my-gcp-project.iam.gserviceaccount.com",
        "client_id": "123456789",
        "auth_uri": "https://accounts.google.com/o/oauth2/auth",
        "token_uri": "https://oauth2.googleapis.com/token"
      }

High-Volume​

Optimizing for high-throughput message consumption...

- id: 3
  name: pubsub_high_volume
  type: pubsub
  properties:
    project_id: "my-gcp-project"
    subscription_id: "high-volume-subscription"
    workers: 16
    max_outstanding_messages: 5000
    max_outstanding_bytes: 2000000000
note

Setting max_outstanding_bytes above available memory can cause out-of-memory errors during traffic bursts.

Pipeline Processing​

Applying custom processing to Pub/Sub messages...

- id: 4
  name: pubsub_pipeline
  type: pubsub
  pipelines:
    - json_parser
    - field_extractor
  properties:
    project_id: "my-gcp-project"
    subscription_id: "structured-logs"
note

Pipelines are processed sequentially and can modify or drop messages before ingestion.

Memory-Constrained​

Limiting memory usage in resource-constrained deployments...

- id: 5
  name: pubsub_low_memory
  type: pubsub
  properties:
    project_id: "my-gcp-project"
    subscription_id: "my-subscription"
    workers: 2
    max_outstanding_messages: 100
    max_outstanding_bytes: 10000000