apache/beam · error · NotImplementedError

from_runner_api_parameter does not handle empty or None topi

Error message

from_runner_api_parameter does not handle empty or None topic

What it means

When reconstructing a Pub/Sub Write transform from its runner API payload, Write.from_runner_api_parameter requires payload.topic to be non-empty because the underlying _PubSubSink needs a target topic. An empty or None topic raises NotImplementedError. The topic cannot be defaulted at deserialization time.

Source

Thrown at sdks/python/apache_beam/io/iobase.py:1162

    else:
      return super().to_runner_api_parameter(context)

  @staticmethod
  @ptransform.PTransform.register_urn(
      common_urns.composites.PUBSUB_WRITE.urn,
      beam_runner_api_pb2.PubSubWritePayload)
  def from_runner_api_parameter(
      ptransform: Any,
      payload: beam_runner_api_pb2.PubSubWritePayload,
      unused_context: PipelineContext,
  ) -> "Write":
    if ptransform.spec.urn != common_urns.composites.PUBSUB_WRITE.urn:
      raise ValueError(
          'Write transform cannot be constructed for the given proto %r',
          ptransform)

    if not payload.topic:
      raise NotImplementedError(
          "from_runner_api_parameter does not "
          "handle empty or None topic")

    # Importing locally to prevent circular dependencies.
    from apache_beam.io.gcp.pubsub import _PubSubSink
    sink = _PubSubSink(
        topic=payload.topic,
        id_label=payload.id_attribute or None,
        timestamp_attribute=payload.timestamp_attribute or None)
    return Write(sink)


class WriteImpl(ptransform.PTransform):
  """Implements the writing of custom sinks."""
  def __init__(self, sink: Sink) -> None:
    super().__init__()
    self.sink = sink

View on GitHub (pinned to 12126d8942)

Solutions

1. Always pass a valid topic (projects/<project>/topics/<name>) when building the Pub/Sub write transform.
2. Check that the component producing the PubSubWritePayload actually populates the topic field before serialization.
3. Upgrade/align Beam versions where payload serialization of topic may have been fixed.
4. If deserializing, validate payload.topic is non-empty before calling from_runner_api_parameter.

Example fix

# before
Write(urn, payload_builder)  # payload has no topic set
# after
payload = PubSubWritePayload(topic='projects/my-project/topics/my-topic', ...)

if not payload.topic:
    raise ValueError('PubSubWritePayload requires a non-empty topic')

def has_topic(payload) -> bool:
    return bool(getattr(payload, 'topic', None))

try:
    write = Write.from_runner_api_parameter(ptransform, payload, ctx)
except NotImplementedError as e:
    if 'topic' in str(e): log.error('Payload missing topic: %r', payload)

When it happens

Trigger: Deserializing a PUBSUB_WRITE transform whose PubSubWritePayload has topic set to '' or None - e.g. the transform was serialized from a pipeline built without a topic, or the topic field was stripped by an intermediate service.

Common situations: Programmatic pipeline construction with WriteToPubSub built via raw runner API protos; expansion services dropping the topic attribute; constructing the write composite directly with no topic argument.

Understand the failure class

Background: "is required", "must be set", "missing required field": configuration validation errors across open-source libraries — this error's family across 36 libraries.

Related errors

• Write transform cannot be constructed for the given proto %r
• One of topic or subscription may be specified.
• Unknown type tag %x
• Expected 0 or 1, got %s
• Could not find enum descriptor: {full_name}

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