apache/beam · error · TimeoutError

PubSub publish timeout exceeded {self.FLUSH_TIMEOUT_SECS} se

Error message

PubSub publish timeout exceeded {self.FLUSH_TIMEOUT_SECS} seconds

What it means

_PubSubSink's _flush waits on all buffered publish futures with a deadline of FLUSH_TIMEOUT_SECS; if any future cannot complete before the remaining budget reaches zero, it raises TimeoutError, aborting the bundle flush.

Source

Thrown at sdks/python/apache_beam/io/gcp/pubsub.py:696

    for elem in self._buffer:
      # Deserialize the protobuf to get the original PubsubMessage
      pubsub_msg = PubsubMessage._from_proto_str(elem)

      # Publish with the correct data, attributes, and ordering_key
      kwargs = {}
      if self.with_attributes and pubsub_msg.attributes:
        kwargs.update(pubsub_msg.attributes)
      if pubsub_msg.ordering_key:
        kwargs['ordering_key'] = pubsub_msg.ordering_key
      future = self._pub_client.publish(self._topic, pubsub_msg.data, **kwargs)

      futures.append(future)

    timer_start = time.time()
    for future in futures:
      remaining = self.FLUSH_TIMEOUT_SECS - (time.time() - timer_start)
      if remaining <= 0:
        raise TimeoutError(
            f"PubSub publish timeout exceeded {self.FLUSH_TIMEOUT_SECS} seconds"
        )
      future.result(remaining)
    self._buffer = []


class _PubSubSink(object):
  """Sink for a Cloud Pub/Sub topic.

  This sink works for both streaming and batch pipelines by using a DoFn
  that buffers and batches messages for efficient publishing.
  """
  def __init__(
      self,
      topic: str,
      id_label: Optional[str],
      timestamp_attribute: Optional[str],
  ):

View on GitHub (pinned to 12126d8942)

Solutions

Reduce the number of messages buffered per bundle (flush earlier in process()) so the timeout budget per future is larger
Increase throughput headroom: request Pub/Sub quota increase or publish to a regional endpoint closer to workers
Add retry with exponential backoff for transient publish failures and monitor network latency; re-run failed bundles

Example fix

// before
future.result()  # waits forever / or long cumulative waits
// after
batch = self._buffer[:MAX_BATCH]
for f in self._publish(batch):
    f.result(timeout=self.FLUSH_TIMEOUT_SECS)  # bounded per-publish, smaller batches

Defensive patterns

Strategy: retry

Validate before calling

assert len(self._buffer) <= MAX_BATCH, 'flush more often to stay under publish timeout'

Try / catch

try:
    future.result(remaining)
except TimeoutError:
    log.warning('PubSub publish timed out; retrying pending messages')
    retry_with_backoff(self._publish, messages)
finally:
    self._buffer = []

Prevention

Flush smaller batches more frequently to bound per-future wait time
Monitor Pub/Sub quota usage and request increases before throttling occurs
Verify worker network access to pubsub.googleapis.com and prefer co-located regions
Enable retries on the underlying google-cloud-pubsub publisher client

When it happens

Trigger: Pub/Sub publishing slow or blocked (quota throttling, network latency to googleapis.com, very large batches) so publish futures do not resolve within FLUSH_TIMEOUT_SECS; long pauses at future.result() exhausting the shared timer budget across many futures.

Common situations: High-throughput batch pipelines exceeding publish quotas; network instability or DNS issues in the worker environment; publishing many messages in a single DoFn bundle so the cumulative wait exceeds the timeout.

Understand the failure class

Background: Request timed out: what client-side request timeouts mean across libraries (Request timed out, TIMED_OUT, APITimeoutError) — this error's family across 39 libraries.

Timeouts: ETIMEDOUT, deadlines, and hung requests — what actually expires when a request times out.

Related errors

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/082ffded0924d144. Report an issue: GitHub.