apache/beam · error · ValueError

PubSub source descriptor must be in the form "projects/

Error message

PubSub source descriptor must be in the form "projects/<project>/topics/<topic>" or "projects/<project>/subscription/<subscription>" (got %r).

What it means

apache_beam raises this ValueError in _WriteToPubSub.__init__ when a PubSub source descriptor string does not match the required pattern projects/<project>/topics/<topic> or projects/<project>/subscriptions/<subscription>. Beam validates the descriptor with PUBSUB_DESCRIPTOR_REGEXP before building the pipeline so malformed topic/subscription paths fail fast at construction time rather than at runtime on the runner.

Solutions

1. Prefix the topic with 'projects/<your-project>/topics/' (e.g. 'projects/my-proj/topics/my-topic').
2. For subscriptions use 'projects/<your-project>/subscriptions/<sub>' (note the plural 'subscriptions').
3. Strip any scheme/host from console URLs and keep only the projects/... resource path.
4. Verify the string matches ^projects/[^/]+/(topics|subscriptions)/[^/]+$ before constructing the transform.

Example fix

// before
beam.io.WriteToPubSub('my-topic')
// after
beam.io.WriteToPubSub('projects/my-gcp-project/topics/my-topic')

import re
PUBSUB_DESCRIPTOR_REGEXP = re.compile(r'projects/[a-zA-Z0-9-_.~]+/(topics|subscriptions)/[a-zA-Z0-9-_.~%+]+')
def is_valid_descriptor(d: str) -> bool:
    return bool(PUBSUB_DESCRIPTOR_REGEXP.match(d))
assert is_valid_descriptor('projects/my-proj/topics/my-topic')

def is_pubsub_descriptor(x: object) -> bool:
    return isinstance(x, str) and bool(PUBSUB_DESCRIPTOR_REGEXP.match(x))

try:
    beam.io.WriteToPubSub(topic)
except ValueError as e:
    if 'PubSub source descriptor' in str(e):
        topic = f'projects/{project}/topics/{topic}'
    else:
        raise

When it happens

Trigger: Passing a string like 'my-topic', 'projects/p/topics/', or a full https://pubsub.googleapis.com/... URL as the topic/subscription argument to WriteToPubSub / ReadFromPubSub (or constructing ReadFromPubSub with an invalid source).

Common situations: Users pass just the topic name without the projects/<project>/topics/ prefix; copy-pasted full resource URLs from the Cloud Console; typos in 'topics'/'subscriptions' (singular vs plural); project IDs containing characters outside the allowed pattern.

Understand the failure class

Background: "Invalid ... format", "must be in format X", "does not look like a ..." — invalid argument format errors across CLI tools and libraries — this error's family across 17 libraries.

Related errors

• An unsupported sink was specified
• At least one of --render_port or --render_output must be…
• buffer_sec must be >= 0, got
• Cannot skip negative number of header lines
• change_function must be 'CHANGES' or 'APPENDS', got

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

Appendix: source

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

      with_attributes: bool = False,
  ):
    """Initializes ``PubSubMultipleReader``.

    Args:
      pubsub_source_descriptors: List of Cloud Pub/Sub topics or subscriptions
        of type `~PubSubSourceDescriptor`.
      with_attributes:
        True - input elements will be :class:`~PubsubMessage` objects.
        False - input elements will be of type ``bytes`` (message data only).
    """
    self.pubsub_source_descriptors = pubsub_source_descriptors
    self.with_attributes = with_attributes

    for descriptor in self.pubsub_source_descriptors:
      match_descriptor = re.match(PUBSUB_DESCRIPTOR_REGEXP, descriptor.source)

      if not match_descriptor:
        raise ValueError(
            'PubSub source descriptor must be in the form "projects/<project>'
            '/topics/<topic>" or "projects/<project>/subscription'
            '/<subscription>" (got %r).' % descriptor.source)

  def expand(self, pcol):
    sources_pcol = []
    for descriptor in self.pubsub_source_descriptors:
      source_match = re.match(PUBSUB_DESCRIPTOR_REGEXP, descriptor.source)
      source_project = source_match.group(1)
      source_type = source_match.group(2)
      source_name = source_match.group(3)

      read_step_name = 'PubSub %s/project:%s/Read %s' % (
          source_type, source_project, source_name)

      if source_type == 'topics':
        current_source = pcol | read_step_name >> ReadFromPubSub(
            topic=descriptor.source,

View on GitHub (pinned to 12126d8942)