{"record":{"id":"ba499bb8463aac7a","repo":"apache/beam","slug":"requires-stateful-processing-beam-2687","errorCode":null,"errorMessage":"Requires stateful processing (BEAM-2687)","messagePattern":"Requires stateful processing \\(BEAM-2687\\)","errorType":"exception","errorClass":"NotImplementedError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/transforms/util.py","lineNumber":1115,"sourceCode":"        target_batch_overhead=target_batch_overhead,\n        target_batch_duration_secs=target_batch_duration_secs,\n        target_batch_duration_secs_including_fixed_cost=(\n            target_batch_duration_secs_including_fixed_cost),\n        variance=variance,\n        clock=clock,\n        record_metrics=record_metrics)\n    self._element_size_fn = element_size_fn\n    self._max_batch_dur = max_batch_duration_secs\n    self._clock = clock\n    self._length_fn = length_fn\n    if length_fn is not None and bucket_boundaries is None:\n      self._bucket_boundaries = self._DEFAULT_BUCKET_BOUNDARIES\n    else:\n      self._bucket_boundaries = bucket_boundaries\n\n  def expand(self, pcoll):\n    if getattr(pcoll.pipeline.runner, 'is_streaming', False):\n      raise NotImplementedError(\"Requires stateful processing (BEAM-2687)\")\n    elif self._max_batch_dur is not None:\n      coder = coders.registry.get_coder(pcoll)\n      if self._length_fn is not None:\n        keying_dofn = WithLengthBucketKey(\n            self._length_fn, self._bucket_boundaries)\n      else:\n        keying_dofn = WithSharedKey()\n      return pcoll | ParDo(keying_dofn) | ParDo(\n          _pardo_stateful_batch_elements(\n              coder,\n              self._batch_size_estimator,\n              self._max_batch_dur,\n              self._clock))\n    elif pcoll.windowing.is_default():\n      # This is the same logic as _GlobalWindowsBatchingDoFn, but optimized\n      # for that simpler case.\n      return pcoll | ParDo(\n          _GlobalWindowsBatchingDoFn(","sourceCodeStart":1097,"sourceCodeEnd":1133,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/transforms/util.py#L1097-L1133","documentation":"BatchReassignWindows (an internal batching transform in apache_beam.transforms.util) rejects pipelines running on a streaming runner, because grouping into batches requires stateful processing support (tracked under JIRA BEAM-2687). The check happens in expand() when the runner reports is_streaming. The library throws NotImplementedError because the operation simply cannot execute on the given runner.","triggerScenarios":"Applying this batching transform to a PCollection in a pipeline whose runner has is_streaming=True (e.g. Flink/Spark/Beam StreamingDataflow without stateful processing), so the transform's expand() immediately raises before any data flows.","commonSituations":"Developers take a batch-oriented pipeline that works with DirectRunner and switch runner to a streaming job; or they build a pipeline parameterized by runner and accidentally route the batching stage into a streaming context.","solutions":["Run the pipeline with a batch runner (e.g. DirectRunner, DataflowRunner batch mode) for this stage","Use a streaming-native batching/windowing alternative (e.g. GroupIntoBatches with streaming support or fixed windows + GBK)","Check the pipeline options: remove is_streaming settings or move the transform into a separate batch job"],"exampleFix":"// before\nwith beam.Pipeline(options=streaming_options) as p:\n  _ = (p | beam.Create(data) | util.BatchReassignWindows())\n// after\n# Use GroupIntoBatches, which supports streaming\n_ = (p | beam.KVToTuple() | util.GroupIntoBatches(batch_size=100))","handlingStrategy":"validation","validationCode":"if getattr(pipeline.runner, 'is_streaming', False):\n    raise RuntimeError('BatchReassignWindows requires a batch runner')","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Check runner.is_streaming before applying batching transforms","Use GroupIntoBatches for streaming-compatible batching","Keep batch-only stages in batch pipeline jobs"],"tags":["python","apache-beam","streaming","not-implemented","runner-compatibility"],"backgroundTag":"unsupported-operation","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"}