kaxil opened a new pull request, #68800: URL: https://github.com/apache/airflow/pull/68800
When the API server serves logs for a **RUNNING** task instance, `FileTaskHandler` constructs the configured executor to call `get_task_log()` and never starts or shuts it down. `KubernetesExecutor.__init__` eagerly created a `multiprocessing.Manager()`, which forks a `serve_forever` child process. Because the executor instance is cached per process, this leaks **one orphaned `Manager` process (~350-400 MB resident) per API-server worker**. With gunicorn/uvicorn worker recycling, each refreshed worker's `Manager` reparents to PID 1 and is never reaped, so they accumulate and push the API server toward OOM. `get_task_log()` only needs the kube client and the pod namespace; it never touches `task_queue`/`result_queue`/`_manager`. The Manager is purely a scheduling-loop resource. Closes #68693. ## Fix Create the `multiprocessing.Manager()` and its queues lazily in `start()` instead of `__init__()`. The scheduling loop (`start`/`execute_async`/`sync`/`end`) is their only consumer, and the scheduler always calls `start()` first. `end()` now no-ops when the executor was never started. Constructing the executor purely to read logs no longer spawns a Manager. ## Design rationale - **Why `start()` and not a lazy property?** The queues are only used after the scheduler calls `start()`; nothing touches them between construction and `start()`. Tying creation to `start()` keeps the lifecycle explicit and pairs cleanly with `end()`. - **Consistent with `LocalExecutor`.** `LocalExecutor` already defers its queue/process creation to `start()`, so it never leaked on this path. This change brings `KubernetesExecutor` in line with that pattern, which also covers `LocalKubernetesExecutor` (its embedded `KubernetesExecutor` is constructed in `__init__` and only started via `LocalKubernetesExecutor.start()`). ## Affected path - `FileTaskHandler._read()` calls the executor's `get_task_log` for running tasks: [file_task_handler.py#L637](https://github.com/apache/airflow/blob/be44ad1ac6/airflow-core/src/airflow/utils/log/file_task_handler.py#L637) - which loads/instantiates the default executor: [executor_loader.py#L292](https://github.com/apache/airflow/blob/be44ad1ac6/airflow-core/src/airflow/executors/executor_loader.py#L292) ## Backward compatibility No public API change. `task_queue`/`result_queue`/`_manager` are `None` until `start()` is called; all internal consumers run after `start()`. Behavior for a started executor (the scheduler) is unchanged. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
