{"record":{"id":"9a967af4cbabba57","repo":"apache/beam","slug":"write-transform-cannot-be-constructed-for-the-given-proto-r","errorCode":null,"errorMessage":"Write transform cannot be constructed for the given proto %r","messagePattern":"Write transform cannot be constructed for the given proto %r","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/io/iobase.py","lineNumber":1157,"sourceCode":"      payload = beam_runner_api_pb2.PubSubWritePayload(\n          topic=self.sink.full_topic,\n          id_attribute=self.sink.id_label,\n          timestamp_attribute=self.sink.timestamp_attribute)\n      return (common_urns.composites.PUBSUB_WRITE.urn, payload)\n    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):","sourceCodeStart":1139,"sourceCodeEnd":1175,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/io/iobase.py#L1139-L1175","documentation":"Write.from_runner_api_parameter reconstructs a Pub/Sub Write transform from its runner API proto. It only supports protos whose spec URN is composites.PUBSUB_WRITE; any other transform proto raises ValueError. This is an internal deserialization guard exercised when a pipeline is deserialized (e.g. portable/runner-side expansion), not a typical user-facing construction path.","triggerScenarios":"Deserializing a pipeline component whose PTransform proto carries a payload for this writer but has a spec URN other than beams:composites:pubsub_write:v1 - typically a malformed or hand-crafted runner API pipeline proto, or a version/URN mismatch.","commonSituations":"Portable pipeline deserialization with mismatched Beam SDK versions; corrupted or manually edited pipeline JSON/proto; custom runners replaying protos they altered.","solutions":["Verify the pipeline proto was produced by the same Beam SDK version and that the transform URN is composites.PUBSUB_WRITE.","Re-generate the pipeline proto from a fresh pipeline construction instead of reusing a cached/saved pipeline definition.","Align SDK versions between the pipeline authoring environment and the expansion/deserialization service.","If custom, implement/lookup the correct PTransformProvider for your URN."],"exampleFix":"// before: reusing a stale serialized pipeline proto from Beam 2.20 against a 2.50 runner service\n// after: rebuild the pipeline with matching versions, or pin both sides to the same apache-beam version","handlingStrategy":"validation","validationCode":"if transform.spec.urn != 'beam:composites:pubsub_write:v1':\n    raise ValueError('unsupported URN: %s' % transform.spec.urn)","typeGuard":"def is_pubsub_write_proto(ptransform) -> bool:\n    return getattr(getattr(ptransform, 'spec', None), 'urn', None) == 'beam:composites:pubsub_write:v1'","tryCatchPattern":"try:\n    write = Write.from_runner_api_parameter(ptransform, payload, ctx)\nexcept ValueError as e:\n    log.error('Cannot deserialize transform %s: %s', ptransform.spec.urn, e)","preventionTips":["Pin identical apache-beam versions across SDK and runner/expansion services","Never hand-edit serialized pipeline protos","Regenerate pipeline protos after upgrades"],"tags":["apache-beam","python","pubsub","deserialization","portable-runner"],"backgroundTag":"unexpected-response-shape","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"}