From b68d1935f958a480f1e7d8dc1e3415707a14646b Mon Sep 17 00:00:00 2001
From: "Leah E. Cole" <6719667+leahecole@users.noreply.github.com>
Date: Thu, 19 Aug 2021 01:50:53 -0700
Subject: [PATCH] Add error check for config_file parameter in
 GKEStartPodOperator (#17700)

---
 .../google/cloud/operators/kubernetes_engine.py     |  4 ++++
 .../cloud/operators/test_kubernetes_engine.py       | 13 +++++++++++++
 2 files changed, 17 insertions(+)

diff --git a/airflow/providers/google/cloud/operators/kubernetes_engine.py b/airflow/providers/google/cloud/operators/kubernetes_engine.py
index c763fb0d621c0..578836d13be90 100644
--- a/airflow/providers/google/cloud/operators/kubernetes_engine.py
+++ b/airflow/providers/google/cloud/operators/kubernetes_engine.py
@@ -292,6 +292,10 @@ def __init__(
                 "Credentials (ADC) strategy for authorization, create an empty connection "
                 "called `google_cloud_default`.",
             )
+        # There is no need to manage the kube_config file, as it will be generated automatically.
+        # All Kubernetes parameters (except config_file) are also valid for the GKEStartPodOperator.
+        if self.config_file:
+            raise AirflowException("config_file is not an allowed parameter for the GKEStartPodOperator.")
 
     def execute(self, context) -> Optional[str]:
         hook = GoogleBaseHook(gcp_conn_id=self.gcp_conn_id)
diff --git a/tests/providers/google/cloud/operators/test_kubernetes_engine.py b/tests/providers/google/cloud/operators/test_kubernetes_engine.py
index a0288df8058d9..0326e7c72cd8e 100644
--- a/tests/providers/google/cloud/operators/test_kubernetes_engine.py
+++ b/tests/providers/google/cloud/operators/test_kubernetes_engine.py
@@ -182,6 +182,19 @@ def test_execute(self, file_mock, mock_execute_in_subprocess, mock_gcp_hook, exe
 
         assert self.gke_op.config_file == FILE_NAME
 
+    def test_config_file_throws_error(self):
+        with pytest.raises(AirflowException):
+            GKEStartPodOperator(
+                project_id=TEST_GCP_PROJECT_ID,
+                location=PROJECT_LOCATION,
+                cluster_name=CLUSTER_NAME,
+                task_id=PROJECT_TASK_ID,
+                name=TASK_NAME,
+                namespace=NAMESPACE,
+                image=IMAGE,
+                config_file="/path/to/alternative/kubeconfig",
+            )
+
     @mock.patch.dict(os.environ, {})
     @mock.patch(
         "airflow.hooks.base.BaseHook.get_connections",