apache/beam · error · TypeError

UnboundedSource.split() produced %r, expected…

Error message

UnboundedSource.split() produced %r, expected UnboundedSource

What it means

The split() entry point of the UnboundedSource restriction provider calls source.split() and validates that every returned sub-source is an UnboundedSource. A non-UnboundedSource result violates the source API contract, so it fails loudly with TypeError rather than silently dropping or mis-tracking the split. This indicates the custom source's split() implementation is buggy.

Solutions

1. Fix the custom source's split() to return only UnboundedSource instances (wrap each result in the correct class, e.g. via dataclasses.replace or a constructor)
2. Unwrap any tuples/containers returned by split() so only sources are yielded
3. Add unit tests calling source.split(N, options) and asserting every item is an UnboundedSource

Example fix

def split(self, desired_num_splits, pipeline_options=None):
    # before: return [(src, state) for src in self._sources]
    # after
    return [src for (src, _state) in self._sources]  # yield only UnboundedSource instances

splits = source.split(n, pipeline_options)
assert all(isinstance(s, UnboundedSource) for s in splits), 'split() returned non-UnboundedSource items'

def valid_splits(xs) -> bool:
    return all(isinstance(x, UnboundedSource) for x in xs)

try:
    list(source.split(n, options))
except TypeError:
    # log and fix the source's split() implementation before running the pipeline
    ...

When it happens

Trigger: A custom UnboundedSource whose split(desired_num_splits, pipeline_options) returns objects that are not UnboundedSource instances (e.g. raw tuples, wrappers, or None entries) during bundle-splitting at pipeline runtime.

Common situations: Implementing a custom UnboundedSource and returning helper/config objects from split(); refactoring split() to return (source, state) tuples and forgetting to unwrap; version changes where split()'s expected return type changed.

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/430805c7fe9b51b9. Report an issue: GitHub.

Appendix: source

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

    try:
      split_sources = list(
          restriction.source.split(_DEFAULT_DESIRED_NUM_SPLITS, None))
    except Exception:  # pylint: disable=broad-except
      _LOGGER.warning(
          'Exception while splitting UnboundedSource. Source not split.',
          exc_info=True)
      yield restriction
      return

    if not split_sources:
      yield restriction
      return

    # A non-UnboundedSource split result is a contract violation, not a
    # refusal, so fail loudly (outside the try/except above).
    for split_source in split_sources:
      if not isinstance(split_source, UnboundedSource):
        raise TypeError(
            'UnboundedSource.split() produced %r, expected UnboundedSource' %
            (split_source, ))

    for split_source in split_sources:
      yield dataclasses.replace(
          restriction,
          source=split_source,
          checkpoint_mark=None,
          is_done=False,
          finalization_checkpoint_mark=None)

  def restriction_size(self, element, restriction) -> int:
    # TODO(https://github.com/apache/beam/issues/19137): implement backlog
    # estimation.
    return 1

  def restriction_coder(self) -> Coder:
    return self._restriction_coder

View on GitHub (pinned to 12126d8942)