apache/beam · error · TypeError

ReadFromUnboundedSource expected an UnboundedSource…

Error message

ReadFromUnboundedSource expected an UnboundedSource element, got %r

What it means

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.

Solutions

1. 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)
2. Check for accidental wrapping/conversion of the source before the transform
3. If using a bounded source, use the bounded read path instead of the unbounded one

Example fix

# before
sources = beam.Create(['my-source'])
_ = sources | ExpandUnboundedSources()
# after
sources = beam.Create([MyUnboundedSource(...)])
_ = sources | ExpandUnboundedSources()

assert isinstance(element, UnboundedSource), 'expected UnboundedSource, got %r' % type(element)

def is_unbounded_source(x) -> bool:
    return isinstance(x, UnboundedSource)

try:
    provider.initial_restriction(element)
except TypeError:
    # route bounded sources to the bounded read path
    ...

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: Type mismatch errors: IllegalArgumentException, TypeError and type guards across 150 open-source libraries — this error's family across 150 libraries.

Related errors

• A cluster_identifier should be Optional[Union[str…
• Cannot get a type descriptor for
• Cannot interpret as Duration.
• CombineGlobally can be used only with combineFn objects…
• database_config must be VectorDatabaseWriteConfig, got

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

Appendix: source

Thrown at sdks/python/apache_beam/io/unbounded_source.py:696

class _UnboundedSourceRestrictionProvider(core.RestrictionProvider):
  """Wraps an :class:`UnboundedSource` element as an SDF restriction.

  Stateless module-level singleton (see :data:`_PROVIDER`): all
  source-specific state (e.g. the source's checkpoint coder) is derived
  per-call from the restriction's ``source`` field, which lets
  :class:`_ReadFromUnboundedSourceDoFn` live at module level too. The provider
  currently passes ``None`` for the ``options`` forwarded to
  :meth:`UnboundedSource.split`.
  """
  def __init__(self):
    self._restriction_coder = _UnboundedSourceRestrictionCoder()

  def initial_restriction(
      self, element: UnboundedSource) -> _UnboundedSourceRestriction:
    if not isinstance(element, UnboundedSource):
      raise TypeError(
          'ReadFromUnboundedSource expected an UnboundedSource element, got %r'
          % (element, ))
    return _UnboundedSourceRestriction(source=element)

  def create_tracker(
      self, restriction: _UnboundedSourceRestriction
  ) -> _UnboundedSourceRestrictionTracker:
    return _UnboundedSourceRestrictionTracker(restriction)

  def split(self, element,
            restriction) -> Iterable[_UnboundedSourceRestriction]:
    if restriction.is_done or restriction.checkpoint_mark is not None:
      yield restriction
      return

    # ``source.split`` is user code and may refuse to split; fall back to a
    # single restriction on error.
    try:

View on GitHub (pinned to 12126d8942)