{"record":{"id":"eae95ca43d2a5d62","repo":"apache/beam","slug":"to-runner-api-parameter-not-implemented-for-type","errorCode":null,"errorMessage":"to_runner_api_parameter not implemented for type","messagePattern":"to_runner_api_parameter not implemented for type","errorType":"exception","errorClass":"NotImplementedError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/io/iobase.py","lineNumber":1010,"sourceCode":"              subscription=self.source.full_subscription,\n              timestamp_attribute=self.source.timestamp_attribute,\n              with_attributes=self.source.with_attributes,\n              id_attribute=self.source.id_label))\n    if isinstance(self.source, BoundedSource):\n      return (\n          common_urns.deprecated_primitives.READ.urn,\n          beam_runner_api_pb2.ReadPayload(\n              source=self.source.to_runner_api(context),\n              is_bounded=beam_runner_api_pb2.IsBounded.BOUNDED\n              if self.source.is_bounded() else\n              beam_runner_api_pb2.IsBounded.UNBOUNDED))\n    # Local import to avoid a circular dependency.\n    from apache_beam.io.unbounded_source import UnboundedSource\n    if isinstance(self.source, UnboundedSource):\n      return super().to_runner_api_parameter(context)\n    elif isinstance(self.source, ptransform.PTransform):\n      return self.source.to_runner_api_parameter(context)\n    raise NotImplementedError(\n        \"to_runner_api_parameter not \"\n        \"implemented for type\")\n\n  @staticmethod\n  def from_runner_api_parameter(\n      transform: beam_runner_api_pb2.PTransform,\n      payload: Union[beam_runner_api_pb2.ReadPayload,\n                     beam_runner_api_pb2.PubSubReadPayload],\n      context: PipelineContext,\n  ) -> \"Read\":\n    if transform.spec.urn == common_urns.composites.PUBSUB_READ.urn:\n      assert isinstance(payload, beam_runner_api_pb2.PubSubReadPayload)\n      # Importing locally to prevent circular dependencies.\n      # TODO(BEAM-27443): Remove the need for this.\n      from apache_beam.io.gcp.pubsub import _PubSubSource\n      source = _PubSubSource(\n          topic=payload.topic or None,\n          subscription=payload.subscription or None,","sourceCodeStart":992,"sourceCodeEnd":1028,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/io/iobase.py#L992-L1028","documentation":"When Beam serializes a CustomSource transform to the runner API, to_runner_api_parameter dispatches on the wrapped source type: UnboundedSource goes to the base class and PTransform sources delegate to their own method. If the wrapped source is neither, NotImplementedError 'to_runner_api_parameter not implemented for type' is raised, meaning Beam does not know how to translate that source for the pipeline runner.","triggerScenarios":"Using a custom Source object that is neither an UnboundedSource nor a PTransform as the payload of CustomSource (e.g. a legacy io.Source subclass or a hand-rolled source) on a portable/Flink/Dataflow runner; wrapping a plain object by mistake.","commonSituations":"Migrating old batch Source-based connectors to newer Beam versions with portable pipelines; third-party or in-house sources never updated for the runner API; version mismatch where an expected io class no longer inherits UnboundedSource.","solutions":["Implement to_runner_api_parameter (and from_runner_api_parameter) on your source class, or make it a proper PTransform so delegation works.","Migrate the custom source to a supported BoundedSource/filebasedsource-based implementation or a built-in Beam connector.","Wrap your reading logic as a PTransform and use that as the source.","Check Beam version compatibility of the third-party io module and upgrade to a release supporting the runner API.","Inspect self.source's actual type/class to confirm why it matches neither branch."],"exampleFix":null,"handlingStrategy":"type-guard","validationCode":"from apache_beam.io.iobase import CustomSource\nfrom apache_beam.io.unbounded_source import UnboundedSource\nimport apache_beam as beam\nassert isinstance(src, (UnboundedSource, beam.PTransform)), \\\n    f'unsupported source type for runner API: {type(src)}'","typeGuard":"def is_runner_api_compatible_source(src):\n    from apache_beam.io.unbounded_source import UnboundedSource\n    import apache_beam as beam\n    return isinstance(src, (UnboundedSource, beam.PTransform))","tryCatchPattern":"try:\n    pipeline.run()\nexcept NotImplementedError as e:\n    if 'to_runner_api_parameter' in str(e):\n        raise TypeError(\n            'Custom source is not serializable to the runner API; '\n            'convert it to a PTransform or supported BoundedSource') from e\n    raise","preventionTips":["Avoid wrapping legacy io.Source subclasses in CustomSource for portable runners.","Prefer built-in Beam connectors or filebasedsource-based implementations.","Register custom to_runner_api_parameter/from_runner_api_parameter for custom transforms.","Pin Beam and connector library versions so source classes inherit the expected base types."],"tags":["python","apache-beam","runner-api","serialization"],"backgroundTag":"method-not-implemented","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"}