{"record":{"id":"430805c7fe9b51b9","repo":"apache/beam","slug":"unboundedsource-split-produced-r-expected-unboundedsource","errorCode":null,"errorMessage":"UnboundedSource.split() produced %r, expected UnboundedSource","messagePattern":"UnboundedSource\\.split\\(\\) produced %r, expected UnboundedSource","errorType":"validation","errorClass":"TypeError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/io/unbounded_source.py","lineNumber":732,"sourceCode":"    try:\n      split_sources = list(\n          restriction.source.split(_DEFAULT_DESIRED_NUM_SPLITS, None))\n    except Exception:  # pylint: disable=broad-except\n      _LOGGER.warning(\n          'Exception while splitting UnboundedSource. Source not split.',\n          exc_info=True)\n      yield restriction\n      return\n\n    if not split_sources:\n      yield restriction\n      return\n\n    # A non-UnboundedSource split result is a contract violation, not a\n    # refusal, so fail loudly (outside the try/except above).\n    for split_source in split_sources:\n      if not isinstance(split_source, UnboundedSource):\n        raise TypeError(\n            'UnboundedSource.split() produced %r, expected UnboundedSource' %\n            (split_source, ))\n\n    for split_source in split_sources:\n      yield dataclasses.replace(\n          restriction,\n          source=split_source,\n          checkpoint_mark=None,\n          is_done=False,\n          finalization_checkpoint_mark=None)\n\n  def restriction_size(self, element, restriction) -> int:\n    # TODO(https://github.com/apache/beam/issues/19137): implement backlog\n    # estimation.\n    return 1\n\n  def restriction_coder(self) -> Coder:\n    return self._restriction_coder","sourceCodeStart":714,"sourceCodeEnd":750,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/io/unbounded_source.py#L714-L750","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["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)","Unwrap any tuples/containers returned by split() so only sources are yielded","Add unit tests calling source.split(N, options) and asserting every item is an UnboundedSource"],"exampleFix":"def split(self, desired_num_splits, pipeline_options=None):\n    # before: return [(src, state) for src in self._sources]\n    # after\n    return [src for (src, _state) in self._sources]  # yield only UnboundedSource instances","handlingStrategy":"type-guard","validationCode":"splits = source.split(n, pipeline_options)\nassert all(isinstance(s, UnboundedSource) for s in splits), 'split() returned non-UnboundedSource items'","typeGuard":"def valid_splits(xs) -> bool:\n    return all(isinstance(x, UnboundedSource) for x in xs)","tryCatchPattern":"try:\n    list(source.split(n, options))\nexcept TypeError:\n    # log and fix the source's split() implementation before running the pipeline\n    ...","preventionTips":["Unit-test split() output types for every custom UnboundedSource","Never return tuples or state objects from split(); only UnboundedSource instances","Keep split() and the restriction provider in sync after refactors"],"tags":["python","apache-beam","type-error","custom-source"],"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"}