This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new bdfc02b8ffb Fix GKE operator imports with older Kubernetes providers
(#74505)
bdfc02b8ffb is described below
commit bdfc02b8ffb647da6bc56dfc1e3a563887e7ec33
Author: Alejandro Morgante <[email protected]>
AuthorDate: Fri Oct 9 15:45:15 2026 -0300
Fix GKE operator imports with older Kubernetes providers (#74505)
* Keep GKE operators importable with older Kubernetes providers
Older Kubernetes provider versions lack the Pod exec operator. That
optional feature must not prevent users from importing or creating existing GKE
operators.
* Exclude the GKE compatibility fallback from example coverage
The compatibility placeholder cannot be used independently and therefore
has no system example.
* Chain the pod_exec ImportError and split the fallback test
Keep the original ImportError as the cause of the missing-provider
error, import AirflowOptionalProviderFeatureException at module top,
and split the fallback test so each behaviour is asserted separately.
Generated-by: Claude Opus 5
---------
Co-authored-by: Jarek Potiuk <[email protected]>
---
.../tests/unit/always/test_project_structure.py | 2 ++
.../google/cloud/operators/kubernetes_engine.py | 26 ++++++++++-----
.../cloud/operators/test_kubernetes_engine.py | 39 ++++++++++++++++++++--
3 files changed, 55 insertions(+), 12 deletions(-)
diff --git a/airflow-core/tests/unit/always/test_project_structure.py
b/airflow-core/tests/unit/always/test_project_structure.py
index ad6629acb2f..8149b3806d6 100644
--- a/airflow-core/tests/unit/always/test_project_structure.py
+++ b/airflow-core/tests/unit/always/test_project_structure.py
@@ -391,6 +391,8 @@ class
TestGoogleProviderProjectStructure(ExampleCoverageTest, AssetsCoverageTest
"airflow.providers.google.cloud.operators.dataproc._DataprocStartStopClusterBaseOperator",
"airflow.providers.google.cloud.operators.dataplex.DataplexCatalogBaseOperator",
"airflow.providers.google.cloud.operators.managed_kafka.ManagedKafkaBaseOperator",
+ # Compatibility fallback for GKEPodExecOperator, not a standalone
Google operator.
+
"airflow.providers.google.cloud.operators.kubernetes_engine.KubernetesPodExecOperator",
"airflow.providers.google.cloud.operators.vertex_ai.custom_job.CustomTrainingJobBaseOperator",
"airflow.providers.google.cloud.operators.vertex_ai.ray.RayBaseOperator",
"airflow.providers.google.cloud.operators.ray.RayJobBaseOperator",
diff --git
a/providers/google/src/airflow/providers/google/cloud/operators/kubernetes_engine.py
b/providers/google/src/airflow/providers/google/cloud/operators/kubernetes_engine.py
index cfdb01d28d6..26842379936 100644
---
a/providers/google/src/airflow/providers/google/cloud/operators/kubernetes_engine.py
+++
b/providers/google/src/airflow/providers/google/cloud/operators/kubernetes_engine.py
@@ -41,7 +41,13 @@ from airflow.providers.cncf.kubernetes.operators.resource
import (
KubernetesDeleteResourceOperator,
)
from airflow.providers.cncf.kubernetes.utils.pod_manager import OnFinishAction
-from airflow.providers.common.compat.sdk import AirflowException, conf,
timezone
+from airflow.providers.common.compat.sdk import (
+ AirflowException,
+ AirflowOptionalProviderFeatureException,
+ BaseOperator,
+ conf,
+ timezone,
+)
from airflow.providers.google.cloud.hooks.kubernetes_engine import (
GKEHook,
GKEKubernetesHook,
@@ -63,19 +69,21 @@ from airflow.providers_manager import ProvidersManager
try:
from airflow.providers.cncf.kubernetes.operators.pod_exec import
KubernetesPodExecOperator
-except ImportError:
- from airflow.providers.common.compat.sdk import
AirflowOptionalProviderFeatureException
+except ImportError as e:
+ _pod_exec_import_error = e
+
+ class KubernetesPodExecOperator(BaseOperator): # type: ignore[no-redef]
+ """Keep existing GKE operators importable with older Kubernetes
providers."""
+
+ def __init__(self, **kwargs) -> None:
+ raise AirflowOptionalProviderFeatureException(
+ "GKEPodExecOperator requires
apache-airflow-providers-cncf-kubernetes>=10.22.0."
+ ) from _pod_exec_import_error
- raise AirflowOptionalProviderFeatureException(
- "Failed to import KubernetesPodExecOperator. This operator is only
available in cncf-kubernetes "
- "provider version >=10.22.0"
- )
try:
from airflow.providers.cncf.kubernetes.operators.job import
KubernetesDeleteJobOperator
except ImportError:
- from airflow.providers.common.compat.sdk import
AirflowOptionalProviderFeatureException
-
raise AirflowOptionalProviderFeatureException(
"Failed to import KubernetesDeleteJobOperator. This operator is only
available in cncf-kubernetes "
"provider version >=8.1.0"
diff --git
a/providers/google/tests/unit/google/cloud/operators/test_kubernetes_engine.py
b/providers/google/tests/unit/google/cloud/operators/test_kubernetes_engine.py
index 0008fc31c37..f820e493bfb 100644
---
a/providers/google/tests/unit/google/cloud/operators/test_kubernetes_engine.py
+++
b/providers/google/tests/unit/google/cloud/operators/test_kubernetes_engine.py
@@ -829,12 +829,45 @@ class TestGKEStartKueueInsideClusterOperator:
)
[email protected]("sys.modules",
{"airflow.providers.cncf.kubernetes.operators.pod_exec": None})
+def
test_missing_kubernetes_pod_exec_operator_keeps_other_gke_operators_usable():
+ operators = runpy.run_path(kubernetes_engine.__file__)
+ create_cluster = operators["GKECreateClusterOperator"](
+ task_id=TEST_TASK_ID,
+ project_id=TEST_PROJECT_ID,
+ location=TEST_LOCATION,
+ body={"name": GKE_CLUSTER_NAME, "node_pools": [{"name": "pool",
"initial_node_count": 1}]},
+ )
+ start_pod = operators["GKEStartPodOperator"](
+ task_id=TEST_TASK_ID,
+ project_id=TEST_PROJECT_ID,
+ location=TEST_LOCATION,
+ cluster_name=GKE_CLUSTER_NAME,
+ image=TEST_IMAGE,
+ )
+
+ assert create_cluster.location == TEST_LOCATION
+ assert start_pod.cluster_name == GKE_CLUSTER_NAME
+ assert start_pod.image == TEST_IMAGE
+
+
@mock.patch.dict("sys.modules",
{"airflow.providers.cncf.kubernetes.operators.pod_exec": None})
def
test_missing_kubernetes_pod_exec_operator_reports_required_provider_version():
+ operators = runpy.run_path(kubernetes_engine.__file__)
+
with pytest.raises(
- AirflowOptionalProviderFeatureException, match="cncf-kubernetes
provider version >=10.22.0"
- ):
- runpy.run_path(kubernetes_engine.__file__)
+ AirflowOptionalProviderFeatureException,
+ match=r"GKEPodExecOperator requires
apache-airflow-providers-cncf-kubernetes>=10\.22\.0",
+ ) as exc_info:
+ operators["GKEPodExecOperator"](
+ task_id=TEST_TASK_ID,
+ location=TEST_LOCATION,
+ cluster_name=GKE_CLUSTER_NAME,
+ pod_name=K8S_POD_NAME,
+ command=["echo", "hello"],
+ )
+
+ assert isinstance(exc_info.value.__cause__, ImportError)
class TestGKEPodExecOperator:
