{"record":{"id":"0ba3d3074bcd40f3","repo":"apache/beam","slug":"returning-elements-from-subprocessdofn-finish-bundle-not","errorCode":null,"errorMessage":"Returning elements from _SubprocessDoFn.finish_bundle not safe.","messagePattern":"Returning elements from _SubprocessDoFn\\.finish_bundle not safe\\.","errorType":"exception","errorClass":"RuntimeError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/transforms/core.py","lineNumber":2752,"sourceCode":"    if cls._fn is None:\n      cls._fn = pickler.loads(cls._serialized_fn)\n      cls._fn.setup()\n    if not cls._started:\n      cls._fn.start_bundle()\n      cls._started = True\n    result = cls._fn.process(*args, **kwargs)\n    if result:\n      # Don't return generator objects.\n      result = list(result)\n    return result\n\n  @classmethod\n  def _remote_finish_bundle(cls):\n    if cls._started:\n      cls._started = False\n      if cls._fn.finish_bundle():\n        # This is because we restart and re-initialize the pool if it crashed.\n        raise RuntimeError(\n            \"Returning elements from _SubprocessDoFn.finish_bundle not safe.\")\n\n  @classmethod\n  def _remote_teardown(cls):\n    if cls._fn:\n      cls._fn.teardown()\n    cls._fn = None\n\n\nclass _TimeoutDoFn(DoFn):\n  \"\"\"Process method run in a separate thread allowing timeouts.\n  \"\"\"\n  def __init__(self, fn, timeout=None):\n    self._fn = fn\n    self._timeout = timeout\n    self._pool = None\n\n  def __getattribute__(self, name):","sourceCodeStart":2734,"sourceCodeEnd":2770,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/transforms/core.py#L2734-L2770","documentation":"_SubprocessDoFn runs the user DoFn in a subprocess pool that can crash and be restarted at any time. Because elements returned from finish_bundle would be buffered across pool restarts and could be duplicated or lost, Beam forbids returning any elements from finish_bundle in this mode and raises RuntimeError if any are produced.","triggerScenarios":"Using .with_exception_handling(use_subprocess=True) (or otherwise wrapping a DoFn in _SubprocessDoFn) where the wrapped DoFn defines a finish_bundle method that yields/returns bundle output elements.","commonSituations":"Users combining the subprocess isolation feature with a DoFn that emits side-bag results in finish_bundle (e.g. flushing accumulated aggregates at end of bundle), which Beam cannot make safe under pool restarts.","solutions":["Move the finish_bundle output logic into process() so elements are emitted normally","Return None / an empty result from finish_bundle when running under subprocess mode","Split the transform: emit via process, and do finish-time bookkeeping only (no emission) in finish_bundle"],"exampleFix":"// before\nclass MyDoFn(DoFn):\n  def finish_bundle(self):\n    yield self.buffered\n// after\nclass MyDoFn(DoFn):\n  def process(self, element):\n    yield element\n    if self.flush_needed:\n      yield self.buffered\n      self.buffered = []\n  def finish_bundle(self):\n    return None","handlingStrategy":"validation","validationCode":"if use_subprocess and hasattr(fn, 'finish_bundle') and getattr(fn, 'emits_in_finish_bundle', True):\n    raise ValueError('DoFn must not emit elements from finish_bundle under use_subprocess')","typeGuard":"def subprocess_safe(fn) -> bool:\n    fb = getattr(fn, 'finish_bundle', None)\n    return fb is None or not getattr(fn, 'emits_in_finish_bundle', True)","tryCatchPattern":"try:\n    run_pipeline()\nexcept RuntimeError as e:\n    if 'finish_bundle not safe' in str(e):\n        refactor_dofn_to_emit_in_process()\n    else:\n        raise","preventionTips":["Never yield from finish_bundle when using subprocess-isolated exception handling","Document which DoFns are subprocess-safe in your team's transform library","Prefer emitting from process() for any DoFn intended for subprocess mode"],"tags":["python","apache-beam","dofn","subprocess","finish-bundle"],"backgroundTag":"unsupported-operation","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}