{"record":{"id":"d8c41f225756aabc","repo":"apache/beam","slug":"unboundedsource-restriction-was-neither-finished-nor","errorCode":null,"errorMessage":"UnboundedSource restriction was neither finished nor checkpointed; process() must self-checkpoint via defer_remainder() or run to EOF: %r","messagePattern":"UnboundedSource restriction was neither finished nor checkpointed; process\\(\\) must self-checkpoint via defer_remainder\\(\\) or run to EOF: %r","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/io/unbounded_source.py","lineNumber":664,"sourceCode":"        finalization_checkpoint_mark=checkpoint)\n    residual = _UnboundedSourceRestriction(\n        source=self._restriction.source,\n        checkpoint_mark=self._clone_checkpoint(checkpoint),\n        watermark=watermark,\n        is_done=False,\n        finalization_checkpoint_mark=None)\n    self._restriction = primary\n    self._checkpoint_taken = True\n    # Park the reader so the resuming bundle reclaims it; on a cache miss the\n    # residual rebuilds one from its checkpoint mark.\n    self._park_or_close_reader(residual)\n    return primary, residual\n\n  def check_done(self) -> bool:\n    # Called after every process(); must raise if work is left unaccounted for.\n    if self._restriction.is_done or self._checkpoint_taken:\n      return True\n    raise ValueError(\n        'UnboundedSource restriction was neither finished nor checkpointed; '\n        'process() must self-checkpoint via defer_remainder() or run to EOF: '\n        '%r' % (self._restriction, ))\n\n  def current_progress(self) -> 'iobase.RestrictionProgress':\n    # Backlog-based progress is not implemented; report a coarse done/not-done\n    # signal via ``completed`` / ``remaining``.\n    if self._restriction.is_done:\n      return iobase.RestrictionProgress(completed=1.0, remaining=0.0)\n    return iobase.RestrictionProgress(completed=0.0, remaining=1.0)\n\n  def is_bounded(self) -> bool:\n    return False\n\n\nclass _UnboundedSourceRestrictionProvider(core.RestrictionProvider):\n  \"\"\"Wraps an :class:`UnboundedSource` element as an SDF restriction.\n","sourceCodeStart":646,"sourceCodeEnd":682,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/io/unbounded_source.py#L646-L682","documentation":"check_done() is called by the Beam splittable-DoFn framework after every process() call on an unbounded restriction tracker. It raises ValueError when the restriction is neither fully consumed nor has a checkpoint been taken, meaning process() neither ran to EOF nor called defer_remainder()/try_split(). This invariant guarantees every element of the unbounded source is accounted for in either the current bundle or a deferred residual; otherwise elements could be lost or duplicated.","triggerScenarios":"Calling process() on the tracker's source and returning without consuming the whole restriction and without calling defer_remainder() (or try_split) to checkpoint the remainder; custom UnboundedSource restriction trackers that advance the restriction partway then return control without self-checkpointing.","commonSituations":"Writing a custom UnboundedSource or RestrictionTracker for streaming reads; a process() loop with a bug that breaks out early (e.g. on a transient error) without deferring the remaining restriction; porting code that assumed another process() invocation would resume mid-restriction.","solutions":["Make process() either read until the restriction is done or, when stopping early, call tracker.defer_remainder(...) / try_split() to checkpoint the remaining work before returning","If the loop must stop early, ensure the stop path always checkpoints (e.g. wrap early returns so the residual is deferred)","Fix logic errors where the restriction is advanced but not marked done and no checkpoint is recorded","Add tests driving process() with a tracker and asserting check_done() returns True afterwards"],"exampleFix":"def process(self, source_restriction_tracker):\n    # before: return after partial read, no checkpoint\n    #   for item in read_some(): yield item\n    #   return\n    # after: defer the un-read remainder before returning\n    for item in read_some():\n        yield item\n    if not source_restriction_tracker.check_done():\n        source_restriction_tracker.defer_remainder(None)","handlingStrategy":"validation","validationCode":"if not restriction_tracker.is_done() and not restriction_tracker.try_split(float('inf')):\n    raise ValueError('process() must consume the restriction or checkpoint it before returning')","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Always end process() with either a completed restriction or a try_split/defer_remainder call","Test custom unbounded sources with a driver that asserts tracker.check_done() after each process()","Never break out of a read loop without accounting for the un-read remainder"],"tags":["python","apache-beam","streaming","splittable-dofn"],"backgroundTag":"invalid-state-transition","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}