{"record":{"id":"a554d52de45d1198","repo":"apache/beam","slug":"readfromunboundedsource-expected-an-unboundedsource-element","errorCode":null,"errorMessage":"ReadFromUnboundedSource expected an UnboundedSource element, got %r","messagePattern":"ReadFromUnboundedSource expected an UnboundedSource element, got %r","errorType":"validation","errorClass":"TypeError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/io/unbounded_source.py","lineNumber":696,"sourceCode":"\n\nclass _UnboundedSourceRestrictionProvider(core.RestrictionProvider):\n  \"\"\"Wraps an :class:`UnboundedSource` element as an SDF restriction.\n\n  Stateless module-level singleton (see :data:`_PROVIDER`): all\n  source-specific state (e.g. the source's checkpoint coder) is derived\n  per-call from the restriction's ``source`` field, which lets\n  :class:`_ReadFromUnboundedSourceDoFn` live at module level too. The provider\n  currently passes ``None`` for the ``options`` forwarded to\n  :meth:`UnboundedSource.split`.\n  \"\"\"\n  def __init__(self):\n    self._restriction_coder = _UnboundedSourceRestrictionCoder()\n\n  def initial_restriction(\n      self, element: UnboundedSource) -> _UnboundedSourceRestriction:\n    if not isinstance(element, UnboundedSource):\n      raise TypeError(\n          'ReadFromUnboundedSource expected an UnboundedSource element, got %r'\n          % (element, ))\n    return _UnboundedSourceRestriction(source=element)\n\n  def create_tracker(\n      self, restriction: _UnboundedSourceRestriction\n  ) -> _UnboundedSourceRestrictionTracker:\n    return _UnboundedSourceRestrictionTracker(restriction)\n\n  def split(self, element,\n            restriction) -> Iterable[_UnboundedSourceRestriction]:\n    if restriction.is_done or restriction.checkpoint_mark is not None:\n      yield restriction\n      return\n\n    # ``source.split`` is user code and may refuse to split; fall back to a\n    # single restriction on error.\n    try:","sourceCodeStart":678,"sourceCodeEnd":714,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/io/unbounded_source.py#L678-L714","documentation":"initial_restriction() of the UnboundedSource restriction provider requires the PCollection element to be an instance of UnboundedSource. It raises TypeError when a different type is passed, because it cannot construct a _UnboundedSourceRestriction around it. This is a programming/wiring error: the splittable DoFn expansion for ReadFromUnboundedSource was applied to the wrong element type.","triggerScenarios":"Applying the unbounded-source restriction provider (via ExpandUnboundedSources or a similar transform) to a PCollection whose elements are not UnboundedSource instances, e.g. a BoundedSource, a string, or an already-materialized record.","commonSituations":"Mixing bounded and unbounded source APIs (switching ReadFromBoundedSource to unbounded variants); passing a wrapped adapter object instead of the actual UnboundedSource; test harnesses feeding placeholder elements.","solutions":["Ensure the PCollection you apply the transform to contains UnboundedSource instances (e.g. create via beam.Create([my_unbounded_source]) where my_unbounded_source subclasses UnboundedSource)","Check for accidental wrapping/conversion of the source before the transform","If using a bounded source, use the bounded read path instead of the unbounded one"],"exampleFix":"# before\nsources = beam.Create(['my-source'])\n_ = sources | ExpandUnboundedSources()\n# after\nsources = beam.Create([MyUnboundedSource(...)])\n_ = sources | ExpandUnboundedSources()","handlingStrategy":"type-guard","validationCode":"assert isinstance(element, UnboundedSource), 'expected UnboundedSource, got %r' % type(element)","typeGuard":"def is_unbounded_source(x) -> bool:\n    return isinstance(x, UnboundedSource)","tryCatchPattern":"try:\n    provider.initial_restriction(element)\nexcept TypeError:\n    # route bounded sources to the bounded read path\n    ...","preventionTips":["Keep unbounded sources in PCollections fed exclusively to the unbounded expansion transform","Avoid wrapping sources in adapters before expansion","Add isinstance asserts in pipeline-construction helpers"],"tags":["python","apache-beam","type-error","splittable-dofn"],"backgroundTag":"type-mismatch","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}