{"record":{"id":"082ffded0924d144","repo":"apache/beam","slug":"pubsub-publish-timeout-exceeded-self-flush-timeout-secs","errorCode":null,"errorMessage":"PubSub publish timeout exceeded {self.FLUSH_TIMEOUT_SECS} seconds","messagePattern":"PubSub publish timeout exceeded (.+?) seconds","errorType":"exception","errorClass":"TimeoutError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/io/gcp/pubsub.py","lineNumber":696,"sourceCode":"    for elem in self._buffer:\n      # Deserialize the protobuf to get the original PubsubMessage\n      pubsub_msg = PubsubMessage._from_proto_str(elem)\n\n      # Publish with the correct data, attributes, and ordering_key\n      kwargs = {}\n      if self.with_attributes and pubsub_msg.attributes:\n        kwargs.update(pubsub_msg.attributes)\n      if pubsub_msg.ordering_key:\n        kwargs['ordering_key'] = pubsub_msg.ordering_key\n      future = self._pub_client.publish(self._topic, pubsub_msg.data, **kwargs)\n\n      futures.append(future)\n\n    timer_start = time.time()\n    for future in futures:\n      remaining = self.FLUSH_TIMEOUT_SECS - (time.time() - timer_start)\n      if remaining <= 0:\n        raise TimeoutError(\n            f\"PubSub publish timeout exceeded {self.FLUSH_TIMEOUT_SECS} seconds\"\n        )\n      future.result(remaining)\n    self._buffer = []\n\n\nclass _PubSubSink(object):\n  \"\"\"Sink for a Cloud Pub/Sub topic.\n\n  This sink works for both streaming and batch pipelines by using a DoFn\n  that buffers and batches messages for efficient publishing.\n  \"\"\"\n  def __init__(\n      self,\n      topic: str,\n      id_label: Optional[str],\n      timestamp_attribute: Optional[str],\n  ):","sourceCodeStart":678,"sourceCodeEnd":714,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/io/gcp/pubsub.py#L678-L714","documentation":"_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.","triggerScenarios":"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.","commonSituations":"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.","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"],"exampleFix":"// before\nfuture.result()  # waits forever / or long cumulative waits\n// after\nbatch = self._buffer[:MAX_BATCH]\nfor f in self._publish(batch):\n    f.result(timeout=self.FLUSH_TIMEOUT_SECS)  # bounded per-publish, smaller batches","handlingStrategy":"retry","validationCode":"assert len(self._buffer) <= MAX_BATCH, 'flush more often to stay under publish timeout'","typeGuard":null,"tryCatchPattern":"try:\n    future.result(remaining)\nexcept TimeoutError:\n    log.warning('PubSub publish timed out; retrying pending messages')\n    retry_with_backoff(self._publish, messages)\nfinally:\n    self._buffer = []","preventionTips":["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"],"tags":["python","pubsub","timeout","apache-beam"],"backgroundTag":"request-timeout","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}