{"record":{"id":"3b64b16e3ad8e903","repo":"apache/beam","slug":"dofn-self-do-fn-r-has-unsupported-per-key-dofn-param-d-per","errorCode":null,"errorMessage":"DoFn {self.do_fn!r} has unsupported per-key DoFn param {d}. Per-key DoFn params are not yet supported for process_batch (https://github.com/apache/beam/issues/21653).","messagePattern":"DoFn (.+?) has unsupported per-key DoFn param (.+?)\\. Per-key DoFn params are not yet supported for process_batch \\(https://github\\.com/apache/beam/issues/21653\\)\\.","errorType":"exception","errorClass":"NotImplementedError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/common.py","lineNumber":365,"sourceCode":"    self._check_duplicate_dofn_params(self.process_method)\n\n  def _validate_process_batch(self):\n    # type: () -> None\n    self._check_duplicate_dofn_params(self.process_batch_method)\n\n    for d in self.process_batch_method.defaults:\n      if not isinstance(d, core._DoFnParam):\n        continue\n\n      # Helpful errors for params which will be supported in the future\n      if d == (core.DoFn.ElementParam):\n        # We currently assume we can just get the typehint from the first\n        # parameter. ElementParam breaks this assumption\n        raise NotImplementedError(\n            f\"DoFn {self.do_fn!r} uses unsupported DoFn param ElementParam.\")\n\n      if d in (core.DoFn.KeyParam, core.DoFn.StateParam, core.DoFn.TimerParam):\n        raise NotImplementedError(\n            f\"DoFn {self.do_fn!r} has unsupported per-key DoFn param {d}. \"\n            \"Per-key DoFn params are not yet supported for process_batch \"\n            \"(https://github.com/apache/beam/issues/21653).\")\n\n      # Fallback to catch anything not explicitly supported\n      if not d in (core.DoFn.WindowParam,\n                   core.DoFn.TimestampParam,\n                   core.DoFn.PaneInfoParam):\n        raise ValueError(\n            f\"DoFn {self.do_fn!r} has unsupported process_batch \"\n            f\"method parameter {d}\")\n\n  def _validate_bundle_method(self, method_wrapper):\n    \"\"\"Validate that none of the DoFnParameters are used in the function\n    \"\"\"\n    for param in core.DoFn.DoFnProcessParams:\n      if param in method_wrapper.defaults:\n        raise ValueError(","sourceCodeStart":347,"sourceCodeEnd":383,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/common.py#L347-L383","documentation":"process_batch does not yet support per-key DoFn parameters (KeyParam, StateParam, TimerParam); these require the per-element/per-key invocation model. _validate_process_batch raises NotImplementedError with a link to Beam issue 21653 tracking the feature.","triggerScenarios":"Declaring process_batch on a stateful DoFn or with core.DoFn.KeyParam/StateParam/TimerParam defaults; validation runs at DoFnSignature creation (_validate).","commonSituations":"Combining batch processing with stateful/timer logic; converting an existing stateful process() to process_batch hoping for batching performance.","solutions":["Keep process() for stateful/per-key DoFns; batching and state are currently incompatible","Move state/timer access into a separate process() DoFn in the pipeline stage","Track/upgrade against Beam issue 21653 for when support lands"],"exampleFix":"// before\ndef process_batch(self, els, state=DoFn.StateParam(Spec)):\n// after\ndef process(self, el, state=DoFn.StateParam(Spec)):","handlingStrategy":"fallback","validationCode":"import inspect\nfrom apache_beam.transforms.core import DoFn\nPER_KEY = (DoFn.KeyParam, DoFn.StateParam, DoFn.TimerParam)\nbad = [p for p in inspect.signature(MyDoFn.process_batch).values if p.default in PER_KEY]\nassert not bad, 'process_batch cannot use per-key params'","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Keep stateful DoFns on process()","Separate batching from stateful logic into different stages"],"tags":["python","apache-beam","stateful-dofn"],"backgroundTag":"feature-not-enabled","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"}