Skip to main content

gcp_pubsub

Sends messages to a GCP Cloud Pub/Sub topic. Metadata from messages are sent as attributes.

# Common config fields, showing default values
output:
  label: ""
  gcp_pubsub:
    project: ""
    topic: ""
    max_in_flight: 1
    metadata:
      exclude_prefixes: []

For information on how to set up credentials check out this guide.

Troubleshooting​

If you're consistently seeing Failed to send message to gcp_pubsub: context deadline exceeded error logs without any further information it is possible that you are encountering https://github.com/Jeffail/benthos/issues/1042, which occurs when metadata values contain characters that are not valid utf-8. This can frequently occur when consuming from Kafka as the key metadata field may be populated with an arbitrary binary value, but this issue is not exclusive to Kafka.

If you are blocked by this issue then a work around is to delete either the specific problematic keys:

pipeline:
  processors:
    - bloblang: |
        meta kafka_key = deleted()

Or delete all keys with:

pipeline:
  processors:
    - bloblang: meta = deleted()

Performance​

This output benefits from sending multiple messages in flight in parallel for improved performance. You can tune the max number of in flight messages with the field max_in_flight.

Fields​

project​

The project ID of the topic to publish to.

Type: string
Default: ""

topic​

The topic to publish to. This field supports interpolation functions.

Type: string
Default: ""

max_in_flight​

The maximum number of messages to have in flight at a given time. Increase this to improve throughput.

Type: int
Default: 1

publish_timeout​

The maximum length of time to wait before abandoning a publish attempt for a message.

Type: string
Default: "60s" 

# Examples


publish_timeout: 10s


publish_timeout: 5m


publish_timeout: 60m

ordering_key​

The ordering key to use for publishing messages. This field supports interpolation functions.

Type: string
Default: ""

metadata​

Specify criteria for which metadata values are sent as attributes.

Type: object

metadata.exclude_prefixes​

Provide a list of explicit metadata key prefixes to be excluded when adding metadata to sent messages.

Type: array
Default: []