apache/beam · error · ValueError

DoFn is splittable but DoFn does not have a…

Error message

DoFn is splittable but DoFn does not have a RestrictionTrackerParam defined

What it means

Apache Beam raises this when a DoFn is detected as splittable (its process method has a RestrictionParam) but the process method has no parameter annotated with @RestrictionTrackerParam, so the runner has no name under which to pass the RestrictionTrackerView. The restriction_provider_arg_name lookup returned None, meaning the DoFn signature is inconsistent: it requests restrictions but never declares where the tracker should be injected.

Solutions

Add a parameter annotated with core.DoFn.RestrictionTrackerParam(<RestrictionT>) to process()
Verify restriction_provider_arg_name is populated by checking the DoFn signature (DoFnSignature/process_method) for the annotated arg
If the DoFn is not meant to be splittable, remove the RestrictionParam so it is not treated as an SDF

Example fix

// before
def process(self, element, restriction=DoFn.RestrictionParam):
    ...
// after
def process(self, element, restriction=DoFn.RestrictionParam,
            tracker=DoFn.RestrictionTrackerParam(CustomTracker)):
    ...

Defensive patterns

Strategy: validation

Validate before calling

from apache_beam.transforms.core import DoFn

def check_sdf_signature(dofn):
    sig = get_method_signature(dofn.process)
    args = [a for a in sig.args if not a.startswith('_')]
    has_restriction = any('restriction' in a.lower() for a in args)
    has_tracker = any('tracker' in a.lower() for a in args)
    if has_restriction and not has_tracker:
        raise ValueError('SDF needs a @DoFn.RestrictionTrackerParam arg')

Type guard

def is_valid_sdf(dofn):
    src = inspect.getsource(dofn.process)
    return 'RestrictionParam' not in src or 'RestrictionTrackerParam' in src

Prevention

Always pair DoFn.RestrictionParam with DoFn.RestrictionTrackerParam in process()
Write a unit test that invokes the DoFn through DoFnInvoker before submitting to a runner
Copy SDF boilerplate from the official Beam examples

When it happens

Trigger: Defining a DoFn whose process() takes a RestrictionParam but omits a RestrictionTrackerParam argument, then running it through a runner that enables SDF expansion via DoFnInvoker.invoke_process.

Common situations: Hand-written splittable DoFns where the author added RestrictionParam but forgot RestrictionTrackerParam; refactors that renamed or dropped the tracker argument; copying an SDF example and trimming parameters.

Understand the failure class

Background: "missing required argument" and "the following required arguments were not provided": what required-argument errors mean and how to fix them — this error's family across 20 libraries.

Related errors

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

Appendix: source

Thrown at sdks/python/apache_beam/runners/common.py:969

    restriction_tracker = self.invoke_create_tracker(self.restriction)
    watermark_estimator = self.invoke_create_watermark_estimator(
        self.watermark_estimator_state)
    with self.splitting_lock:
      if window_index:
        self.current_window_index = window_index
        if window_index == 0:
          self.stop_window_index = len(windowed_value.windows)
        if window_index == self.stop_window_index:
          return False
      self.threadsafe_restriction_tracker = ThreadsafeRestrictionTracker(
          restriction_tracker)
      self.threadsafe_watermark_estimator = (
          ThreadsafeWatermarkEstimator(watermark_estimator))

    restriction_tracker_param = (
        self.signature.process_method.restriction_provider_arg_name)
    if not restriction_tracker_param:
      raise ValueError(
          'DoFn is splittable but DoFn does not have a '
          'RestrictionTrackerParam defined')
    additional_kwargs[restriction_tracker_param] = (
        RestrictionTrackerView(self.threadsafe_restriction_tracker))
    watermark_param = (
        self.signature.process_method.watermark_estimator_provider_arg_name)
    # When the watermark_estimator is a NoOpWatermarkEstimator, the system
    # will not add watermark_param into the DoFn param list.
    if watermark_param is not None:
      additional_kwargs[watermark_param] = self.threadsafe_watermark_estimator
    return True

  def _invoke_process_per_window(
      self,
      windowed_value,  # type: WindowedValue
      additional_args,
      additional_kwargs,
  ):

View on GitHub (pinned to 12126d8942)