apache/beam · warning · RuntimeError

Errors

Error message

Errors: {}

What it means

After cancellation or a terminal state, _cleanup() runs registered cleanup callbacks; if any of them raised, it aggregates all exceptions and raises a single RuntimeError('Errors: {...}') with repr of each. It signals that job teardown (e.g. stopping message threads) partially failed.

Solutions

1. Read the aggregated reprs in the message to identify which callbacks failed.
2. Ensure the job endpoint is reachable — cleanup usually fails from connection errors; reconnect and re-cancel if needed.
3. Wrap cancel() in try/except since the job may already be terminal.
4. Fix or remove faulty custom cleanup callbacks registered on the result handle.

Example fix

try:
    result.cancel()
except RuntimeError as e:
    print('cleanup issues (job may already be done):', e)

try:
    result.cancel()
except RuntimeError as e:
    # aggregated cleanup callback failures; job may already be terminal
    print('cleanup errors:', e)

When it happens

Trigger: Calling cancel() or reaching a terminal state in _observe_state while one or more registered _cleanup_callbacks raised exceptions.

Common situations: Network drops when closing gRPC streams to the job service during cleanup; exceptions in user-registered cleanup hooks; cancel racing with job completion.

Related errors

• At least one owner must be registered.
• Attempting to emit an element outside of a @ProcessElement…
• BigQuery storage source must be split before reading
• BigQuery test was not shutdown previously. Table is
• Buffer counter not initialized for UUID:

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

Appendix: source

Thrown at sdks/python/apache_beam/runners/portability/portable_runner.py:616

    if on_exit and self._cleanup_callbacks:
      _LOGGER.info(
          'Running cleanup on exit. If your pipeline should continue running, '
          'be sure to use the following syntax:\n'
          '  with Pipeline() as p:\n'
          '    p.apply(..)\n'
          'This ensures that the pipeline finishes before this program exits.')
    callback_exceptions = []
    for callback in self._cleanup_callbacks:
      try:
        callback()
      except Exception as e:
        callback_exceptions.append(e)

    self._cleanup_callbacks = ()
    if callback_exceptions:
      formatted_exceptions = ''.join(
          [f"\n\t{repr(e)}" for e in callback_exceptions])
      raise RuntimeError('Errors: {}'.format(formatted_exceptions))

View on GitHub (pinned to 12126d8942)