apache/beam · error · ValueError

Initialize ThreadsafeRestrictionTracker requiresRestrictionT

Error message

Initialize ThreadsafeRestrictionTracker requiresRestrictionTracker.

What it means

ThreadsafeRestrictionTracker wraps a RestrictionTracker to make it thread-safe for splittable DoFn use. Its constructor type-checks the argument and raises this ValueError if the object is not an instance of apache_beam.io.iobase.RestrictionTracker.

Source

Thrown at sdks/python/apache_beam/runners/sdf_utils.py:60

SplitResultPrimary = NamedTuple(
    'SplitResultPrimary', [('primary_value', WindowedValue)])

SplitResultResidual = NamedTuple(
    'SplitResultResidual',
    [('residual_value', WindowedValue), ('current_watermark', Timestamp),
     ('deferred_timestamp', Optional[Duration])])


class ThreadsafeRestrictionTracker(object):
  """A thread-safe wrapper which wraps a `RestrictionTracker`.

  This wrapper guarantees synchronization of modifying restrictions across
  multi-thread.
  """
  def __init__(self, restriction_tracker: 'RestrictionTracker') -> None:
    from apache_beam.io.iobase import RestrictionTracker
    if not isinstance(restriction_tracker, RestrictionTracker):
      raise ValueError(
          'Initialize ThreadsafeRestrictionTracker requires'
          'RestrictionTracker.')
    self._restriction_tracker = restriction_tracker
    # Records an absolute timestamp when defer_remainder is called.
    self._timestamp = None
    self._lock = threading.RLock()
    self._deferred_residual = None
    self._deferred_timestamp: Optional[Union[Timestamp, Duration]] = None

  def current_restriction(self):
    with self._lock:
      return self._restriction_tracker.current_restriction()

  def try_claim(self, position):
    with self._lock:
      return self._restriction_tracker.try_claim(position)

  def defer_remainder(self, deferred_time=None):

View on GitHub (pinned to 12126d8942)

Solutions

1. Pass a valid RestrictionTracker instance (e.g. OffsetRangeTracker) into ThreadsafeRestrictionTracker
2. Make your custom tracker subclass apache_beam.io.iobase.RestrictionTracker
3. Check that you are not passing None or the restriction object itself

Example fix

// before
tracker = ThreadsafeRestrictionTracker(OffsetRange(0, 100))
// after
from apache_beam.io.iobase import RestrictionTracker
from apache_beam.io.restriction_trackers import OffsetRangeTracker
tracker = ThreadsafeRestrictionTracker(OffsetRangeTracker(OffsetRange(0, 100)))

from apache_beam.io.iobase import RestrictionTracker
isinstance(x, RestrictionTracker)  # check before constructing

from apache_beam.io.iobase import RestrictionTracker
def is_restriction_tracker(x) -> bool:
    return isinstance(x, RestrictionTracker)

try:
    tracker = ThreadsafeRestrictionTracker(x)
except ValueError as e:
    if 'RestrictionTracker' in str(e):
        raise TypeError('must pass a RestrictionTracker') from e
    raise

When it happens

Trigger: Instantiating ThreadsafeRestrictionTracker(x) where x is None, a raw restriction, a custom tracker not subclassing RestrictionTracker, or an object from a different tracker hierarchy.

Common situations: Writing a custom splittable DoFn and passing the restriction instead of the tracker; a custom tracker class that forgets to inherit from RestrictionTracker.

Related errors

• Initialize RestrictionTrackerView requires ThreadsafeRestric
• The timestamp of deter_remainder() should be a Duration or a
• Initializing Threadsafe requires a WatermarkEstimator
• Input of observe_timestamp should be a Timestamp object
• DoFn terminated without fully processing restriction

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