apache/beam · error · RuntimeError

Cannot schedule new tasks after thread pool has been shutdow

Error message

Cannot schedule new tasks after thread pool has been shutdown.

What it means

Beam's ThreadPoolExecutor subclass refuses submit() after shutdown() was called: the internal _shutdown flag is checked under the lock before dispatching work to idle workers. Python's concurrent.futures queues the task anyway; this implementation raises eagerly to surface the misuse. This is Beam's own vendored-compatible executor (a copy in this module), not the stdlib ThreadPoolExecutor, which is why the error surfaces here.

Source

Thrown at sdks/python/apache_beam/utils/thread_pool_executor.py:91

class UnboundedThreadPoolExecutor(_base.Executor):
  def __init__(self):
    self._idle_worker_queue = queue.Queue()
    self._max_idle_threads = 16
    self._workers = weakref.WeakSet()
    self._shutdown = False
    self._lock = threading.Lock()  # Guards access to _workers and _shutdown

  def submit(self, fn, *args, **kwargs):
    """Attempts to submit the work item.

    A runtime error is raised if the pool has been shutdown.
    """
    future = _base.Future()
    work_item = _WorkItem(future, fn, args, kwargs)
    with self._lock:
      if self._shutdown:
        raise RuntimeError(
            'Cannot schedule new tasks after thread pool has been shutdown.')
      try:
        self._idle_worker_queue.get(block=False).assign_work(work_item)

        # If we have more idle threads then the max allowed, shutdown a thread.
        if self._idle_worker_queue.qsize() > self._max_idle_threads:
          try:
            self._idle_worker_queue.get(block=False).shutdown()
          except queue.Empty:
            pass
      except queue.Empty:
        worker = _Worker(self._idle_worker_queue, work_item)
        worker.daemon = True
        worker.start()
        self._workers.add(worker)
    return future

  def shutdown(self, wait=True):

View on GitHub (pinned to 12126d8942)

Solutions

Create a fresh ThreadPoolExecutor instead of reusing the shut-down instance.
Keep the pool alive until all submits are done: submit within the same 'with' block or before calling shutdown().
If you need reuse, remove the shutdown()/context-exit between phases and shut down only at final teardown.
Guard teardown ordering so atexit/interpreter shutdown doesn't precede final submits.

Example fix

# before
pool = ThreadPoolExecutor(4)
with pool:
    pool.submit(f)
pool.submit(g)  # RuntimeError
# after
with ThreadPoolExecutor(4) as pool:
    pool.submit(f)
    pool.submit(g)

Defensive patterns

Strategy: type-guard

Validate before calling

def can_submit(pool):
    return not getattr(pool, '_shutdown', True)
if can_submit(pool):
    pool.submit(task)

Type guard

def is_active_pool(pool):
    return isinstance(pool, ThreadPoolExecutor) and not pool._shutdown

Try / catch

try:
    pool.submit(task)
except RuntimeError as e:
    if 'thread pool has been shutdown' in str(e):
        pool = ThreadPoolExecutor(max_workers)
        pool.submit(task)
    else:
        raise

Prevention

Submit all work inside the pool's 'with' block
Don't store pools in module globals that outlive pipeline runs
Only call shutdown() at final teardown
Avoid submitting from destructors/atexit handlers

When it happens

Trigger: Calling submit(fn, ...) on a thread_pool_executor.ThreadPoolExecutor after shutdown() / exiting its context manager / interpreter teardown during which atexit shuts pools down.

Common situations: Long-lived pool objects used across pipeline runs where one run's cleanup shut the pool down; submitting background work from a destructor or atexit handler; reusing a module-level pool after an earlier 'with' block.

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

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