apache/beam · error · RuntimeError

Unclosed error handler initialized at %s

Error message

Unclosed error handler initialized at %s

What it means

At the end of pipeline construction Beam calls verify_closed() on every ErrorHandler to make sure none was created and then abandoned with errors silently dropped. If the handler was never closed (never used as a context manager / never had its output consumed into the pipeline), this RuntimeError fires, pointing at the traceback captured when the handler was created.

Source

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

    """Returns result of applying the error consumer to the error pcollections.
    """
    if not self._closed:
      raise RuntimeError(
          "Cannot access the output of an error handler "
          "until it has been closed.")
    return self._output

  def add_error_pcollection(self, pcoll):
    """Called by a class implementing error handling on the error records.
    """
    pcoll.pipeline._register_error_handler(self)
    self._error_pcolls.append(pcoll)

  def verify_closed(self):
    """Called at end of pipeline construction to ensure errors are not ignored.
    """
    if not self._closed:
      raise RuntimeError(
          "Unclosed error handler initialized at %s" % self._creation_traceback)


class _IdentityPTransform(transforms.PTransform):
  def expand(self, pcoll):
    return pcoll


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

View on GitHub (pinned to 12126d8942)

Solutions

1. Use the handler as a context manager (with error_handling(...) as handler:) so it closes automatically.
2. Check the 'initialized at' traceback in the message to locate the unclosed handler and either use or delete it.
3. If errors are intentionally ignored, wrap the transform differently or explicitly consume/close the handler.

Example fix

# before
handler = RecordCountingErrorHandler()
pcoll.with_exception_handling(handler, ...)  # never closed
# after
with error_handling(RecordCountingErrorHandler()) as handler:
  pcoll.with_exception_handling(handler, ...)

assert getattr(handler, '_closed', False), 'error handler must be used as a context manager'

def is_closed(handler) -> bool:
    return bool(getattr(handler, '_closed', False))

try:
    pipeline.run()
except RuntimeError as e:
    if 'Unclosed error handler' in str(e):
        locate_and_fix_handler(e)  # message includes creation traceback
    raise

When it happens

Trigger: Creating an ErrorHandler (e.g. CollectingErrorHandler or via with_exception_handling) but never exiting its context manager, or constructing one manually and attaching it to no transform before the pipeline is built.

Common situations: Abandoned with-blocks exited early via return/raise, handlers instantiated in interactive sessions for experimentation, or code refactors that removed the with_exception_handling usage but left the handler construction.

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
• CollectingErrorHandler requires the output to be retrieved.
• PCollection used directly as side input argument. Specify As

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