roshanprabu opened a new pull request, #71344:
URL: https://github.com/apache/airflow/pull/71344
## Summary
Fixes #68827.
In-cluster, kubernetes-client 36.x's default `Configuration` (set by
`InClusterConfigLoader`) carries a `refresh_api_key_hook` local closure. Every
openapi-generated model's `__init__` calls `Configuration.get_default_copy()`
when `local_vars_configuration` isn't explicitly passed:
```python
def __init__(self, ..., local_vars_configuration=None):
if local_vars_configuration is None:
local_vars_configuration = Configuration.get_default_copy()
self.local_vars_configuration = local_vars_configuration
```
A task's `pod_override` (a `V1Pod` built by user DAG code, which never
passes `local_vars_configuration`) therefore inherits that closure in-cluster.
`execute_async` embeds it in a `KubernetesJob` and puts it on
`self.task_queue`, a `multiprocessing.Manager().JoinableQueue()` -- the manager
pickles the object to send it over IPC, and `pickle` can't serialize a closure.
This crashes the **scheduler process itself**, not just the task, and since the
task gets re-queued on restart, it loops.
The fix wires up protection that already exists in this codebase but wasn't
used at this call site: `PodGenerator.deserialize_model_dict` builds `V1Pod`
objects against a fresh, empty `Configuration()` specifically so neither the
pod nor any nested model captures the process-global in-cluster default -- its
docstring says as much verbatim. `execute_async` never routed `pod_override`
through it before constructing `KubernetesJob`. This PR rounds it through
`PodGenerator.serialize_pod` + `deserialize_model_dict` right after
`PodGenerator.from_obj`, so the queued object is always built from a fresh
`Configuration`.
**Scope note:** the linked issue also mentions the same fix would help
`OAuth2.py`-style curl callers elsewhere and suggests a broader
`CURLOPT_PROTOCOLS` pass -- not applicable here, that's from a different
(Appwrite) advisory I worked on separately. This PR only touches the
`execute_async` → `task_queue.put()` path described in #68827.
## Test plan
- [x] Reproduced the exact crash locally: constructed a `Configuration` with
an unpicklable `refresh_api_key_hook` (mirroring `InClusterConfigLoader`), set
it as the process default, built a `V1Pod` the way user code would, and
confirmed `pickle.dumps()` fails with `AttributeError: Can't get local object
...` -- then confirmed `PodGenerator.serialize_pod` + `deserialize_model_dict`
produces a pod that pickles/unpickles cleanly with the pod spec intact.
- [x] Added `test_execute_async_pod_override_is_picklable_in_cluster`, which
reproduces the same in-cluster `Configuration` and calls the *real*
`execute_async` → `task_queue.put()` path (the actual production
`multiprocessing.Manager` IPC pickling, not a synthetic `pickle.dumps()` check).
- [x] Verified the new test actually catches the bug: reverted the source
fix and confirmed the test fails at `self.task_queue.put(job)` with the exact
same `AttributeError`; re-applied the fix and confirmed it passes.
- [x] Ran the full `test_kubernetes_executor.py` file: **177 passed, 1
skipped** (pre-existing, version-gated skip unrelated to this change) -- no
regressions.
- [x] `ruff check` and `ruff format --check` pass on both changed files.
--
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]
