This is an automated email from the ASF dual-hosted git repository. dimberman pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/master by this push: new 94d3ed6 Fix error with quick-failing tasks in KubernetesPodOperator (#13621) 94d3ed6 is described below commit 94d3ed61d60b134d649a4e9785b2d9c2a88cff05 Author: Daniel Imberman <daniel.imber...@gmail.com> AuthorDate: Thu Jan 21 12:57:35 2021 -0800 Fix error with quick-failing tasks in KubernetesPodOperator (#13621) * Fix error with quick-failing tasks in KubernetesPodOperator Addresses an issue with the KubernetesPodOperator where tasks that die quickly are not patched with "already_checked" because they never make it to the monitoring logic. * static fix --- .../cncf/kubernetes/operators/kubernetes_pod.py | 1 + kubernetes_tests/test_kubernetes_pod_operator.py | 25 ++++++++++++++++++++++ 2 files changed, 26 insertions(+) diff --git a/airflow/providers/cncf/kubernetes/operators/kubernetes_pod.py b/airflow/providers/cncf/kubernetes/operators/kubernetes_pod.py index 3f42ab1..996b7da 100644 --- a/airflow/providers/cncf/kubernetes/operators/kubernetes_pod.py +++ b/airflow/providers/cncf/kubernetes/operators/kubernetes_pod.py @@ -343,6 +343,7 @@ class KubernetesPodOperator(BaseOperator): # pylint: disable=too-many-instance- raise AirflowException(f'Pod {self.pod.metadata.name} returned a failure: {status}') return result except AirflowException as ex: + self.patch_already_checked(self.pod) raise AirflowException(f'Pod Launching failed: {ex}') def handle_pod_overlap( diff --git a/kubernetes_tests/test_kubernetes_pod_operator.py b/kubernetes_tests/test_kubernetes_pod_operator.py index 574f7ff..e860303 100644 --- a/kubernetes_tests/test_kubernetes_pod_operator.py +++ b/kubernetes_tests/test_kubernetes_pod_operator.py @@ -1000,5 +1000,30 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase): k.execute(context) create_mock.assert_called_once() + def test_reatttach_quick_failure(self): + client = kube_client.get_kube_client(in_cluster=False) + namespace = "default" + + name = "test" + k = KubernetesPodOperator( + namespace='default', + image="ubuntu:16.04", + cmds=["bash", "-cx"], + arguments=["exit 1"], + labels={"foo": "bar"}, + name="test", + task_id=name, + in_cluster=False, + do_xcom_push=False, + is_delete_operator_pod=False, + termination_grace_period=0, + ) + + context = create_context(k) + with self.assertRaises(AirflowException): + k.execute(context) + pod = client.read_namespaced_pod(name=k.pod.metadata.name, namespace=namespace) + self.assertEqual(pod.metadata.labels["already_checked"], "True") + # pylint: enable=unused-argument