apache/beam · error · RuntimeError

Returning elements from _SubprocessDoFn.finish_bundle not sa

Error message

Returning elements from _SubprocessDoFn.finish_bundle not safe.

What it means

_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.

Source

Thrown at sdks/python/apache_beam/transforms/core.py:2752

    if cls._fn is None:
      cls._fn = pickler.loads(cls._serialized_fn)
      cls._fn.setup()
    if not cls._started:
      cls._fn.start_bundle()
      cls._started = True
    result = cls._fn.process(*args, **kwargs)
    if result:
      # Don't return generator objects.
      result = list(result)
    return result

  @classmethod
  def _remote_finish_bundle(cls):
    if cls._started:
      cls._started = False
      if cls._fn.finish_bundle():
        # This is because we restart and re-initialize the pool if it crashed.
        raise RuntimeError(
            "Returning elements from _SubprocessDoFn.finish_bundle not safe.")

  @classmethod
  def _remote_teardown(cls):
    if cls._fn:
      cls._fn.teardown()
    cls._fn = None


class _TimeoutDoFn(DoFn):
  """Process method run in a separate thread allowing timeouts.
  """
  def __init__(self, fn, timeout=None):
    self._fn = fn
    self._timeout = timeout
    self._pool = None

  def __getattribute__(self, name):

View on GitHub (pinned to 12126d8942)

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

Example fix

// before
class MyDoFn(DoFn):
  def finish_bundle(self):
    yield self.buffered
// after
class MyDoFn(DoFn):
  def process(self, element):
    yield element
    if self.flush_needed:
      yield self.buffered
      self.buffered = []
  def finish_bundle(self):
    return None

Defensive patterns

Strategy: validation

Validate before calling

if use_subprocess and hasattr(fn, 'finish_bundle') and getattr(fn, 'emits_in_finish_bundle', True):
    raise ValueError('DoFn must not emit elements from finish_bundle under use_subprocess')

Type guard

def subprocess_safe(fn) -> bool:
    fb = getattr(fn, 'finish_bundle', None)
    return fb is None or not getattr(fn, 'emits_in_finish_bundle', True)

Try / catch

try:
    run_pipeline()
except RuntimeError as e:
    if 'finish_bundle not safe' in str(e):
        refactor_dofn_to_emit_in_process()
    else:
        raise

Prevention

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

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: UnsupportedOperationException and "is not supported" errors: when a library deliberately refuses a call — this error's family across 30 libraries.

Related errors

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/0ba3d3074bcd40f3. Report an issue: GitHub.