apache/beam · error · TypeError

One of topic or subscription may be specified.

Error message

One of topic or subscription may be specified.

What it means

read_from_pubsub got neither 'topic' nor 'subscription'; there is no Pub/Sub endpoint to attach the source to, so the read cannot be constructed. This is the empty-side companion to the topic/subscription mutual-exclusion check.

Source

Thrown at sdks/python/apache_beam/yaml/yaml_io.py:390

      case, deduplication of the stream will be strictly best effort.
    timestamp_attribute: Message value to use as element timestamp. If None,
      uses message publishing time as the timestamp.

      Timestamp values should be in one of two formats:

      - A numerical value representing the number of milliseconds since the
        Unix epoch.
      - A string in RFC 3339 format, UTC timezone. Example:
        ``2015-10-29T23:41:41.123Z``. The sub-second component of the
        timestamp is optional, and digits beyond the first three (i.e., time
        units smaller than milliseconds) may be ignored.
    publish_time_field: Field to add to output messages with the Pub/Sub
      message publish time. If None, no such field is added.
  """
  if topic and subscription:
    raise TypeError('Only one of topic and subscription may be specified.')
  elif not topic and not subscription:
    raise TypeError('One of topic or subscription may be specified.')
  if publish_time_field is not None and not publish_time_field.strip():
    raise ValueError('publish_time_field must be a non-empty field name.')
  has_publish_time_field = publish_time_field is not None
  payload_schema, parser = _create_parser(format, schema)
  extra_fields: list[schema_pb2.Field] = []
  if not attributes and not attributes_map and not has_publish_time_field:
    mapper = lambda msg: parser(msg)
  else:
    if isinstance(attributes, str):
      attributes = [attributes]
    if attributes:
      extra_fields.extend(
          [schemas.schema_field(attr, str) for attr in attributes])
    if attributes_map:
      extra_fields.append(
          schemas.schema_field(attributes_map, Mapping[str, str]))
    if has_publish_time_field:
      extra_fields.append(

View on GitHub (pinned to 12126d8942)

Solutions

1. Pass topic='projects/<proj>/topics/<name>'
2. Or pass subscription='projects/<proj>/subscriptions/<name>'
3. Verify your config/template substitutes the topic value before invocation

Example fix

// before
read_from_pubsub(format='JSON')
// after
read_from_pubsub(topic='projects/my-proj/topics/my-topic', format='JSON')

if not topic and not subscription:
    raise ValueError('Provide either topic or subscription')

def has_read_source(args):
    return bool(args.get('topic') or args.get('subscription'))

try:
    read_from_pubsub(topic=cfg.get('topic'), subscription=cfg.get('subscription'))
except TypeError as e:
    if 'One of topic' in str(e):
        log.error('Configure a Pub/Sub topic or subscription')
        raise

When it happens

Trigger: Calling read_from_pubsub() with both topic=None and subscription=None, e.g. empty YAML keys resolved to None.

Common situations: YAML config omits the topic key or a templated variable is empty/unset, silently yielding None.

Understand the failure class

Background: "missing required argument" and "the following required arguments were not provided": what required-argument errors mean and how to fix them — this error's family across 20 libraries.

Related errors

• from_runner_api_parameter does not handle empty or None topi
• Either data (%r) or attributes (%r) must be set.
• A pubsub message data field must not exceed 10MB
• A pubsub message must not have more than 100 attributes.
• A pubsub message attribute key must not exceed 256 bytes.

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