diogosilva30 opened a new pull request, #65943: URL: https://github.com/apache/airflow/pull/65943
## What Replace `multiprocessing.Process` (fork) with `subprocess.Popen` (fresh interpreter) in the edge3 worker's `_launch_job()` method. Fixes #65942 ## Why The edge worker runs **22+ threads** (asyncio event loop, `ThreadPoolExecutor`, HTTP clients, heartbeat loops). When `_launch_job()` calls `multiprocessing.Process(target=self._run_job_via_supervisor)`, the default `fork` start method copies the entire process — including **locked import locks** held by other threads — into the child. Since only the forking thread survives in the child, those locks are never released, causing: 1. **Permanent deadlocks** — child hangs on any `import` that needs a lock held by a now-dead thread 2. **Corrupted `sys.modules` state** — child inherits partially-initialized modules, causing `ModuleNotFoundError` cascades for all plugin and DAG imports Both failure modes are **intermittent** (~5% of forks in our testing, higher under load) and produce no useful error messages — the task simply times out or exits with code 1. This is a well-known POSIX constraint: [fork() in multi-threaded programs is unsafe](https://docs.python.org/3/library/os.html#os.fork). Python 3.14 will change the default multiprocessing start method to `forkserver` precisely because of this class of bug — but that would cause `PicklingError` in edge3 since `_run_job_via_supervisor` is a bound method on `EdgeWorker` which carries unpicklable state. ## How The fix uses infrastructure that **already exists in the codebase**: 1. `ExecuteTask` is a **Pydantic model** with `model_dump_json()` — the edge executor already serializes it to JSON when storing to the DB 2. `airflow.sdk.execution_time.execute_workload` is an **existing CLI entrypoint** that deserializes an `ExecuteTask` from JSON and calls `supervise()` 3. The **ECS executor already uses this exact pattern**: `["python", "-m", "airflow.sdk.execution_time.execute_workload", "--json-string", workload.model_dump_json()]` ### Changes | File | What changed | |------|-------------| | `worker.py` | `_launch_job()` now uses `subprocess.Popen` with `execute_workload --json-string`. Removed `_run_job_via_supervisor()`, `_reset_parent_signal_state()`, `multiprocessing` imports, and `results_queue` plumbing | | `dataclasses.py` | `Job.process` type changed from `multiprocessing.Process` to `subprocess.Popen`. `is_running` uses `poll() is None`, `is_success` checks `returncode == 0` | | `test_worker.py` | Updated mocks and assertions to match subprocess-based approach | ### What's removed - `_run_job_via_supervisor()` — the `execute_workload` module does the same thing (calls `supervise()` with the deserialized workload) - `_reset_parent_signal_state()` — no longer needed since child is a fresh process, not a fork - `multiprocessing.Process` / `multiprocessing.Queue` imports — replaced by `subprocess.Popen` - `results_queue` error propagation — errors are now detected via `process.returncode != 0` ### Trade-offs | Concern | Assessment | |---------|-----------| | **Startup cost** | ~1-2s per task for new interpreter + DAG parse. Same cost Celery/ECS/K8s executors pay. Acceptable for edge workers where tasks typically run minutes+ | | **Error detail** | `results_queue` gave the full exception traceback. Now we get exit code + whatever the task wrote to its log file. The log file already contains the full traceback via `supervise()` | | **Diff size** | ~30 lines removed, ~15 lines added in production code | ## Testing - Verified on live edge worker pod (RKE cluster, 9 replicas, Python 3.12.13) with 22 active threads - Reproduction script confirms ~95% deadlock rate with `fork`, 0% with `subprocess.Popen` - Existing `execute_workload` entrypoint has its own test suite -- 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]
