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: []