apache/beam · error · ValueError

UnboundedSource restriction was neither finished nor checkpo

Error message

UnboundedSource restriction was neither finished nor checkpointed; process() must self-checkpoint via defer_remainder() or run to EOF: %r

What it means

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.

Source

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

        finalization_checkpoint_mark=checkpoint)
    residual = _UnboundedSourceRestriction(
        source=self._restriction.source,
        checkpoint_mark=self._clone_checkpoint(checkpoint),
        watermark=watermark,
        is_done=False,
        finalization_checkpoint_mark=None)
    self._restriction = primary
    self._checkpoint_taken = True
    # Park the reader so the resuming bundle reclaims it; on a cache miss the
    # residual rebuilds one from its checkpoint mark.
    self._park_or_close_reader(residual)
    return primary, residual

  def check_done(self) -> bool:
    # Called after every process(); must raise if work is left unaccounted for.
    if self._restriction.is_done or self._checkpoint_taken:
      return True
    raise ValueError(
        'UnboundedSource restriction was neither finished nor checkpointed; '
        'process() must self-checkpoint via defer_remainder() or run to EOF: '
        '%r' % (self._restriction, ))

  def current_progress(self) -> 'iobase.RestrictionProgress':
    # Backlog-based progress is not implemented; report a coarse done/not-done
    # signal via ``completed`` / ``remaining``.
    if self._restriction.is_done:
      return iobase.RestrictionProgress(completed=1.0, remaining=0.0)
    return iobase.RestrictionProgress(completed=0.0, remaining=1.0)

  def is_bounded(self) -> bool:
    return False


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

View on GitHub (pinned to 12126d8942)

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

Example fix

def process(self, source_restriction_tracker):
    # before: return after partial read, no checkpoint
    #   for item in read_some(): yield item
    #   return
    # after: defer the un-read remainder before returning
    for item in read_some():
        yield item
    if not source_restriction_tracker.check_done():
        source_restriction_tracker.defer_remainder(None)

Defensive patterns

Strategy: validation

Validate before calling

if not restriction_tracker.is_done() and not restriction_tracker.try_split(float('inf')):
    raise ValueError('process() must consume the restriction or checkpoint it before returning')

Prevention

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

When it happens

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

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

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

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