{"record":{"id":"6eeb24c48d726882","repo":"apache/beam","slug":"cannot-schedule-new-tasks-after-thread-pool-has-been","errorCode":null,"errorMessage":"Cannot schedule new tasks after thread pool has been shutdown.","messagePattern":"Cannot schedule new tasks after thread pool has been shutdown\\.","errorType":"exception","errorClass":"RuntimeError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/utils/thread_pool_executor.py","lineNumber":91,"sourceCode":"\nclass UnboundedThreadPoolExecutor(_base.Executor):\n  def __init__(self):\n    self._idle_worker_queue = queue.Queue()\n    self._max_idle_threads = 16\n    self._workers = weakref.WeakSet()\n    self._shutdown = False\n    self._lock = threading.Lock()  # Guards access to _workers and _shutdown\n\n  def submit(self, fn, *args, **kwargs):\n    \"\"\"Attempts to submit the work item.\n\n    A runtime error is raised if the pool has been shutdown.\n    \"\"\"\n    future = _base.Future()\n    work_item = _WorkItem(future, fn, args, kwargs)\n    with self._lock:\n      if self._shutdown:\n        raise RuntimeError(\n            'Cannot schedule new tasks after thread pool has been shutdown.')\n      try:\n        self._idle_worker_queue.get(block=False).assign_work(work_item)\n\n        # If we have more idle threads then the max allowed, shutdown a thread.\n        if self._idle_worker_queue.qsize() > self._max_idle_threads:\n          try:\n            self._idle_worker_queue.get(block=False).shutdown()\n          except queue.Empty:\n            pass\n      except queue.Empty:\n        worker = _Worker(self._idle_worker_queue, work_item)\n        worker.daemon = True\n        worker.start()\n        self._workers.add(worker)\n    return future\n\n  def shutdown(self, wait=True):","sourceCodeStart":73,"sourceCodeEnd":109,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/utils/thread_pool_executor.py#L73-L109","documentation":"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.","triggerScenarios":"Calling submit(fn, ...) on a thread_pool_executor.ThreadPoolExecutor after shutdown() / exiting its context manager / interpreter teardown during which atexit shuts pools down.","commonSituations":"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.","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."],"exampleFix":"# before\npool = ThreadPoolExecutor(4)\nwith pool:\n    pool.submit(f)\npool.submit(g)  # RuntimeError\n# after\nwith ThreadPoolExecutor(4) as pool:\n    pool.submit(f)\n    pool.submit(g)","handlingStrategy":"type-guard","validationCode":"def can_submit(pool):\n    return not getattr(pool, '_shutdown', True)\nif can_submit(pool):\n    pool.submit(task)","typeGuard":"def is_active_pool(pool):\n    return isinstance(pool, ThreadPoolExecutor) and not pool._shutdown","tryCatchPattern":"try:\n    pool.submit(task)\nexcept RuntimeError as e:\n    if 'thread pool has been shutdown' in str(e):\n        pool = ThreadPoolExecutor(max_workers)\n        pool.submit(task)\n    else:\n        raise","preventionTips":["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"],"tags":["python","concurrency","thread-pool","lifecycle","apache-beam"],"backgroundTag":"invalid-state-transition","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}