{"record":{"id":"1a316a0c1e135116","repo":"apache/beam","slug":"https-github-com-apache-beam-issues-21653-per-key-process","errorCode":null,"errorMessage":"https://github.com/apache/beam/issues/21653: Per-key process_batch","messagePattern":"https://github\\.com/apache/beam/issues/21653: Per-key process_batch","errorType":"exception","errorClass":"NotImplementedError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/common.py","lineNumber":1121,"sourceCode":"      side_inputs = [si[window] for si in self.side_inputs]\n      side_inputs.extend(additional_args)\n      args_for_process_batch, kwargs_for_process_batch = (\n          util.insert_values_in_args(\n              self.args_for_process_batch,\n              self.kwargs_for_process_batch,\n              side_inputs,\n          )\n      )\n      if not self.recalculate_window_args:\n        self.args_for_process_batch, self.kwargs_for_process_batch = (\n            args_for_process_batch, kwargs_for_process_batch)\n        self.has_cached_window_batch_args = True\n\n    for i, p in self.placeholders_for_process_batch:\n      if core.DoFn.ElementParam == p:\n        args_for_process_batch[i] = windowed_batch.values\n      elif core.DoFn.KeyParam == p:\n        raise NotImplementedError(\n            'https://github.com/apache/beam/issues/21653: Per-key process_batch'\n        )\n      elif core.DoFn.WindowParam == p:\n        args_for_process_batch[i] = window\n      elif core.DoFn.TimestampParam == p:\n        args_for_process_batch[i] = windowed_batch.timestamp\n      elif core.DoFn.PaneInfoParam == p:\n        assert isinstance(windowed_batch, HomogeneousWindowedBatch)\n        args_for_process_batch[i] = windowed_batch.pane_info\n      elif isinstance(p, core.DoFn.StateParam):\n        raise NotImplementedError(\n            \"https://github.com/apache/beam/issues/21653: \"\n            \"Per-key process_batch\")\n      elif isinstance(p, core.DoFn.TimerParam):\n        raise NotImplementedError(\n            \"https://github.com/apache/beam/issues/21653: \"\n            \"Per-key process_batch\")\n      elif isinstance(p, core.DoFn.BundleContextParam):","sourceCodeStart":1103,"sourceCodeEnd":1139,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/common.py#L1103-L1139","documentation":"Beam's batched SDF processing (process_batch) does not yet support a per-key KeyParam injection, so a DoFn whose process_batch declares KeyParam raises this NotImplementedError referencing beam issue 21653. This is an intentional, unimplemented-feature guard, not a data error.","triggerScenarios":"Declaring core.DoFn.KeyParam as a parameter of a process_batch method and invoking it via DoFnInvoker.invoke_process_batch on windowed batches.","commonSituations":"Porting an existing per-element process() (with KeyParam) to process_batch for performance; using batched stateful DoFns on runners that route through the batched invoker.","solutions":["Remove KeyParam from process_batch and derive the key inside the method from the elements (they are KVs)","Fall back to per-element process() until BEAM-21653 is implemented","Pre-group/key elements explicitly and use process() with StateParam if per-key access is essential"],"exampleFix":"# before\ndef process_batch(self, batch, key=DoFn.KeyParam):\n    ...\n# after\ndef process_batch(self, batch):\n    for k, v in batch.values:\n        ...  # extract key from each KV element","handlingStrategy":"fallback","validationCode":"import inspect\n\ndef batch_supports_key(dofn):\n    src = inspect.getsource(dofn.process_batch)\n    return 'DoFn.KeyParam' not in src","typeGuard":null,"tryCatchPattern":"try:\n    invoker.invoke_process_batch(batch)\nexcept NotImplementedError as e:\n    if '21653' in str(e):\n        logging.warning('process_batch lacks KeyParam support; falling back to process()')\n        for wb in batch.split():\n            invoker.invoke_process(wb)","preventionTips":["Do not use KeyParam in process_batch until BEAM-21653 lands","Extract keys from KV elements inside process_batch instead","Track the Beam issue for feature availability before migrating"],"tags":["apache-beam","python","process-batch","not-implemented"],"backgroundTag":"method-not-implemented","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"}