{"record":{"id":"5ad747cac43892c3","repo":"apache/beam","slug":"from-runner-api-parameter-does-not-handle-empty-or-none","errorCode":null,"errorMessage":"from_runner_api_parameter does not handle empty or None topic","messagePattern":"from_runner_api_parameter does not handle empty or None topic","errorType":"exception","errorClass":"NotImplementedError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/io/iobase.py","lineNumber":1162,"sourceCode":"    else:\n      return super().to_runner_api_parameter(context)\n\n  @staticmethod\n  @ptransform.PTransform.register_urn(\n      common_urns.composites.PUBSUB_WRITE.urn,\n      beam_runner_api_pb2.PubSubWritePayload)\n  def from_runner_api_parameter(\n      ptransform: Any,\n      payload: beam_runner_api_pb2.PubSubWritePayload,\n      unused_context: PipelineContext,\n  ) -> \"Write\":\n    if ptransform.spec.urn != common_urns.composites.PUBSUB_WRITE.urn:\n      raise ValueError(\n          'Write transform cannot be constructed for the given proto %r',\n          ptransform)\n\n    if not payload.topic:\n      raise NotImplementedError(\n          \"from_runner_api_parameter does not \"\n          \"handle empty or None topic\")\n\n    # Importing locally to prevent circular dependencies.\n    from apache_beam.io.gcp.pubsub import _PubSubSink\n    sink = _PubSubSink(\n        topic=payload.topic,\n        id_label=payload.id_attribute or None,\n        timestamp_attribute=payload.timestamp_attribute or None)\n    return Write(sink)\n\n\nclass WriteImpl(ptransform.PTransform):\n  \"\"\"Implements the writing of custom sinks.\"\"\"\n  def __init__(self, sink: Sink) -> None:\n    super().__init__()\n    self.sink = sink\n","sourceCodeStart":1144,"sourceCodeEnd":1180,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/io/iobase.py#L1144-L1180","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Always pass a valid topic (projects/<project>/topics/<name>) when building the Pub/Sub write transform.","Check that the component producing the PubSubWritePayload actually populates the topic field before serialization.","Upgrade/align Beam versions where payload serialization of topic may have been fixed.","If deserializing, validate payload.topic is non-empty before calling from_runner_api_parameter."],"exampleFix":"# before\nWrite(urn, payload_builder)  # payload has no topic set\n# after\npayload = PubSubWritePayload(topic='projects/my-project/topics/my-topic', ...)","handlingStrategy":"validation","validationCode":"if not payload.topic:\n    raise ValueError('PubSubWritePayload requires a non-empty topic')","typeGuard":"def has_topic(payload) -> bool:\n    return bool(getattr(payload, 'topic', None))","tryCatchPattern":"try:\n    write = Write.from_runner_api_parameter(ptransform, payload, ctx)\nexcept NotImplementedError as e:\n    if 'topic' in str(e): log.error('Payload missing topic: %r', payload)","preventionTips":["Always supply a fully-qualified topic when building Pub/Sub writes","Validate payload fields before serializing/deserializing","Test pipeline round-trips (serialize then deserialize) in CI"],"tags":["apache-beam","python","pubsub","missing-required-argument","deserialization"],"backgroundTag":"missing-required-config-field","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"}