{"record":{"id":"cc2689b90da9a368","repo":"apache/beam","slug":"dofn-is-splittable-but-dofn-does-not-have-a","errorCode":null,"errorMessage":"DoFn is splittable but DoFn does not have a RestrictionTrackerParam defined","messagePattern":"DoFn is splittable but DoFn does not have a RestrictionTrackerParam defined","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/common.py","lineNumber":969,"sourceCode":"    restriction_tracker = self.invoke_create_tracker(self.restriction)\n    watermark_estimator = self.invoke_create_watermark_estimator(\n        self.watermark_estimator_state)\n    with self.splitting_lock:\n      if window_index:\n        self.current_window_index = window_index\n        if window_index == 0:\n          self.stop_window_index = len(windowed_value.windows)\n        if window_index == self.stop_window_index:\n          return False\n      self.threadsafe_restriction_tracker = ThreadsafeRestrictionTracker(\n          restriction_tracker)\n      self.threadsafe_watermark_estimator = (\n          ThreadsafeWatermarkEstimator(watermark_estimator))\n\n    restriction_tracker_param = (\n        self.signature.process_method.restriction_provider_arg_name)\n    if not restriction_tracker_param:\n      raise ValueError(\n          'DoFn is splittable but DoFn does not have a '\n          'RestrictionTrackerParam defined')\n    additional_kwargs[restriction_tracker_param] = (\n        RestrictionTrackerView(self.threadsafe_restriction_tracker))\n    watermark_param = (\n        self.signature.process_method.watermark_estimator_provider_arg_name)\n    # When the watermark_estimator is a NoOpWatermarkEstimator, the system\n    # will not add watermark_param into the DoFn param list.\n    if watermark_param is not None:\n      additional_kwargs[watermark_param] = self.threadsafe_watermark_estimator\n    return True\n\n  def _invoke_process_per_window(\n      self,\n      windowed_value,  # type: WindowedValue\n      additional_args,\n      additional_kwargs,\n  ):","sourceCodeStart":951,"sourceCodeEnd":987,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/common.py#L951-L987","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","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"],"exampleFix":"// before\ndef process(self, element, restriction=DoFn.RestrictionParam):\n    ...\n// after\ndef process(self, element, restriction=DoFn.RestrictionParam,\n            tracker=DoFn.RestrictionTrackerParam(CustomTracker)):\n    ...","handlingStrategy":"validation","validationCode":"from apache_beam.transforms.core import DoFn\n\ndef check_sdf_signature(dofn):\n    sig = get_method_signature(dofn.process)\n    args = [a for a in sig.args if not a.startswith('_')]\n    has_restriction = any('restriction' in a.lower() for a in args)\n    has_tracker = any('tracker' in a.lower() for a in args)\n    if has_restriction and not has_tracker:\n        raise ValueError('SDF needs a @DoFn.RestrictionTrackerParam arg')","typeGuard":"def is_valid_sdf(dofn):\n    src = inspect.getsource(dofn.process)\n    return 'RestrictionParam' not in src or 'RestrictionTrackerParam' in src","tryCatchPattern":null,"preventionTips":["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"],"tags":["apache-beam","python","splittable-dofn"],"backgroundTag":"missing-required-argument","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"}