You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by js...@apache.org on 2026/06/07 12:01:12 UTC
(airflow) branch v3-2-test updated: [v3-2-test] Fix K8s executor pod_override stringified without cncf provider (#67895) (#68157)
This is an automated email from the ASF dual-hosted git repository.
jscheffl pushed a commit to branch v3-2-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-2-test by this push:
new 84565b27fa2 [v3-2-test] Fix K8s executor pod_override stringified without cncf provider (#67895) (#68157)
84565b27fa2 is described below
commit 84565b27fa29912e7d83941164cfa278f7086688
Author: Jens Scheffler <95...@users.noreply.github.com>
AuthorDate: Sun Jun 7 14:01:05 2026 +0200
[v3-2-test] Fix K8s executor pod_override stringified without cncf provider (#67895) (#68157)
* Fix Kubernetes executor pod_override stringified without cncf provider
Airflow 3 deployments using the Kubernetes executor with the `kubernetes`
package installed but without `apache-airflow-providers-cncf-kubernetes`
had their `executor_config` `pod_override` V1Pod silently converted to a
string during Dag serialization. The stringified pod was persisted to
`serialized_dag` and `task_instance.executor_config`, so affected tasks
stayed in `queued` and never ran.
`_has_kubernetes()` hardcoded an import of the cncf provider's
`PodGenerator` for V1Pod ser/deser. When only an alternative executor
package is installed, that import failed, `_has_kubernetes()` cached
False, and serialization fell back to `str(pod)`. Airflow 2 had an
in-core `PodGenerator` fallback for this case that was removed ahead of
the 3.0 release.
Serialize and deserialize V1Pod objects directly through
`kubernetes.client.ApiClient`, which is all `PodGenerator.serialize_pod`
and `deserialize_model_dict` wrap anyway. V1Pod ser/deser now works with
the cncf provider, an alternative executor package, or the bare
`kubernetes` package installed. The backcompat unpickle path in
`ensure_pod_is_valid_after_unpickling` is decoupled the same way.
* Apply suggestions from code review
---------
(cherry picked from commit 4677e1e0c41130676a41379c183efda50adddaca)
Co-authored-by: Ephraim Anierobi <sp...@gmail.com>
Co-authored-by: Copilot Autofix powered by AI <17...@users.noreply.github.com>
---
.../airflow/serialization/serialized_objects.py | 14 ++++-----
airflow-core/src/airflow/utils/sqlalchemy.py | 5 ++--
.../unit/serialization/test_serialized_objects.py | 35 ++++++++++++++++++++--
3 files changed, 43 insertions(+), 11 deletions(-)
diff --git a/airflow-core/src/airflow/serialization/serialized_objects.py b/airflow-core/src/airflow/serialization/serialized_objects.py
index 351a0b6cf42..5b355607877 100644
--- a/airflow-core/src/airflow/serialization/serialized_objects.py
+++ b/airflow-core/src/airflow/serialization/serialized_objects.py
@@ -114,9 +114,9 @@ if TYPE_CHECKING:
from inspect import Parameter
from kubernetes.client import models as k8s # noqa: TC004
+ from kubernetes.client.api_client import ApiClient # noqa: TC004
from airflow.models.expandinput import SchedulerExpandInput
- from airflow.providers.cncf.kubernetes.pod_generator import PodGenerator # noqa: TC004
from airflow.sdk import BaseOperatorLink
from airflow.sdk.definitions._internal.node import DAGNode as SDKDAGNode
from airflow.sdk.types import Operator as SdkOperator
@@ -496,7 +496,7 @@ class BaseSerialization:
and _has_kubernetes(attempt_import=True)
and isinstance(var, k8s.V1Pod)
):
- json_pod = PodGenerator.serialize_pod(var)
+ json_pod = ApiClient().sanitize_for_serialization(var)
return cls._encode(json_pod, type_=DAT.POD)
elif isinstance(var, OutletEventAccessors):
return cls._encode(
@@ -652,9 +652,10 @@ class BaseSerialization:
if not _has_kubernetes(attempt_import=True):
raise RuntimeError(
"Cannot deserialize POD objects without kubernetes libraries. "
- "Please install the cncf.kubernetes provider."
+ "Please install the `kubernetes` package."
)
- pod = PodGenerator.deserialize_model_dict(var)
+ # kubernetes-client does not expose a public dict->model API; see https://github.com/kubernetes-client/python/issues/977.
+ pod = ApiClient()._ApiClient__deserialize_model(var, k8s.V1Pod)
return pod
elif type_ == DAT.TIMEDELTA:
return datetime.timedelta(seconds=var)
@@ -2188,11 +2189,10 @@ def _has_kubernetes(attempt_import: bool = False) -> bool:
# Loading kube modules is expensive, so delay it until the last moment
try:
from kubernetes.client import models as k8s
-
- from airflow.providers.cncf.kubernetes.pod_generator import PodGenerator
+ from kubernetes.client.api_client import ApiClient
globals()["k8s"] = k8s
- globals()["PodGenerator"] = PodGenerator
+ globals()["ApiClient"] = ApiClient
return True
except ImportError:
return False
diff --git a/airflow-core/src/airflow/utils/sqlalchemy.py b/airflow-core/src/airflow/utils/sqlalchemy.py
index 5f9adc99d95..c35edd83aa2 100644
--- a/airflow-core/src/airflow/utils/sqlalchemy.py
+++ b/airflow-core/src/airflow/utils/sqlalchemy.py
@@ -251,11 +251,12 @@ def ensure_pod_is_valid_after_unpickling(pod: V1Pod) -> V1Pod | None:
if not isinstance(pod, V1Pod):
return None
try:
- from airflow.providers.cncf.kubernetes.pod_generator import PodGenerator
+ from kubernetes.client.api_client import ApiClient
# now we actually reserialize / deserialize the pod
pod_dict = sanitize_for_serialization(pod)
- return PodGenerator.deserialize_model_dict(pod_dict)
+ # kubernetes-client does not expose a public dict->model API; see https://github.com/kubernetes-client/python/issues/977.
+ return ApiClient()._ApiClient__deserialize_model(pod_dict, V1Pod)
except Exception:
return None
diff --git a/airflow-core/tests/unit/serialization/test_serialized_objects.py b/airflow-core/tests/unit/serialization/test_serialized_objects.py
index 06754923137..b91ba41c680 100644
--- a/airflow-core/tests/unit/serialization/test_serialized_objects.py
+++ b/airflow-core/tests/unit/serialization/test_serialized_objects.py
@@ -1042,7 +1042,7 @@ class TestKubernetesImportAvoidance:
def test_serialize_v1pod_attempts_import_before_serializing(self, monkeypatch):
"""Regression test: V1Pod serialization must call _has_kubernetes(attempt_import=True)."""
k8s = pytest.importorskip("kubernetes.client.models")
- from airflow.providers.cncf.kubernetes.pod_generator import PodGenerator
+ from kubernetes.client.api_client import ApiClient
calls = []
@@ -1052,7 +1052,7 @@ class TestKubernetesImportAvoidance:
monkeypatch.setattr(serialized_objects, "_has_kubernetes", fake_has_kubernetes)
monkeypatch.setattr(serialized_objects, "k8s", k8s, raising=False)
- monkeypatch.setattr(serialized_objects, "PodGenerator", PodGenerator, raising=False)
+ monkeypatch.setattr(serialized_objects, "ApiClient", ApiClient, raising=False)
pod = k8s.V1Pod(metadata=k8s.V1ObjectMeta(name="test-pod"))
result = BaseSerialization.serialize(pod)
@@ -1060,3 +1060,34 @@ class TestKubernetesImportAvoidance:
assert isinstance(result, dict), "V1Pod should serialize to a dict, not a string"
assert result.get(Encoding.TYPE) == DAT.POD, "V1Pod should have type DAT.POD"
assert True in calls
+
+ def test_v1pod_serde_without_cncf_kubernetes_provider(self, monkeypatch):
+ """V1Pod ser/deser must work when the cncf.kubernetes provider is not installed.
+
+ Regression test for the K8s executor ``pod_override`` getting stringified during DAG
+ serialization on deployments that install ``kubernetes`` but not
+ ``apache-airflow-providers-cncf-kubernetes``.
+ """
+ k8s = pytest.importorskip("kubernetes.client.models")
+
+ # Simulate the cncf.kubernetes provider being unimportable. Setting a module to ``None``
+ # in ``sys.modules`` makes importing it raise ``ModuleNotFoundError``. We must block the
+ # exact module the old code imported (``...pod_generator``) and not just the parent
+ # package: if the submodule is already cached in ``sys.modules`` from an earlier test,
+ # ``from ...pod_generator import PodGenerator`` resolves the cached leaf without ever
+ # touching the parent, so blocking only the parent would not exercise the regression.
+ monkeypatch.setitem(sys.modules, "airflow.providers.cncf.kubernetes", None)
+ monkeypatch.setitem(sys.modules, "airflow.providers.cncf.kubernetes.pod_generator", None)
+ _has_kubernetes.cache_clear()
+
+ pod = k8s.V1Pod(metadata=k8s.V1ObjectMeta(name="test-pod"))
+ encoded = BaseSerialization.serialize(pod)
+
+ assert isinstance(encoded, dict), "V1Pod should serialize to a dict, not a string"
+ assert encoded[Encoding.TYPE] == DAT.POD
+
+ decoded = BaseSerialization.deserialize(encoded)
+ assert isinstance(decoded, k8s.V1Pod)
+ assert decoded.metadata.name == "test-pod"
+
+ _has_kubernetes.cache_clear()