apache/beam · error · RuntimeError

CollectingErrorHandler requires the output to be retrieved.

Error message

CollectingErrorHandler requires the output to be retrieved. Initialized at %s

What it means

CollectingErrorHandler accumulates error records in memory, which are only materialized when output() is called; calling output() also marks the handler as having had its output accessed. verify_closed() raises this RuntimeError if the handler closed without output() ever being read, because the collected error records would otherwise be silently discarded.

Source

Thrown at sdks/python/apache_beam/transforms/error_handling.py:123

class CollectingErrorHandler(ErrorHandler):
  """An ErrorHandler that simply collects all errors for further processing.

  This ErrorHandler requires the set of errors be retrieved via `output()`
  and consumed (or explicitly discarded).
  """
  def __init__(self):
    super().__init__(_IdentityPTransform())
    self._creation_traceback = traceback.format_stack()[-2]
    self._output_accessed = False

  def output(self):
    self._output_accessed = True
    return super().output()

  def verify_closed(self):
    if not self._output_accessed:
      raise RuntimeError(
          "CollectingErrorHandler requires the output to be retrieved. "
          "Initialized at %s" % self._creation_traceback)
    return super().verify_closed()

View on GitHub (pinned to 12126d8942)

Solutions

1. Call handler.output() inside the with-block after applying the transform, and wire the result into the pipeline.
2. Use a different ErrorHandler (e.g. a sink-writing handler) if you don't need to read the collected records.
3. Inspect the 'initialized at' traceback to find the unused handler.

Example fix

# before
with error_handling(CollectingErrorHandler()) as handler:
  pcoll.with_exception_handling(handler, ...)
# after
with error_handling(CollectingErrorHandler()) as handler:
  pcoll.with_exception_handling(handler, ...)
  handler.output()  # consume collected errors

assert getattr(handler, '_output_accessed', False), 'CollectingErrorHandler.output() must be called before close'

def output_consumed(handler) -> bool:
    return bool(getattr(handler, '_output_accessed', False))

try:
    pipeline.run()
except RuntimeError as e:
    if 'requires the output to be retrieved' in str(e):
        add_output_call(e)  # wire handler.output() into the pipeline
    raise

When it happens

Trigger: Using CollectingErrorHandler as a context manager (or closing it) but never calling handler.output() before the pipeline is built.

Common situations: Copying error-handling boilerplate from stateless handlers that don't require output retrieval, or exiting the with-block on an exception path before reading the collected records.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

• A sink must inherit iobase.Sink, iobase.NativeSink, or be a
• Cannot insert different protos %r and %r with the same ID %r
• Cannot access the output of an error handler until it has be
• Unclosed error handler initialized at %s
• PCollection used directly as side input argument. Specify As

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