Copilot commented on code in PR #72606:
URL: https://github.com/apache/airflow/pull/72606#discussion_r3945629636
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py:
##########
@@ -886,6 +886,17 @@ def execute_sync(self, context: Context):
pod=pod_to_clean, remote_pod=self.remote_pod, context=context,
result=result
)
+ if self._killed:
+ # on_kill() ran while the block above was waiting on the pod, and
that wait
+ # returned normally because the pod it was watching simply went
away. The
+ # workload never finished, so falling through here would finalise
the task
+ # instance as success. This check sits after the finally block on
purpose: if
+ # the body raised, that exception propagates untouched and already
fails the
+ # task with its own reason.
+ raise AirflowException(
+ f"Pod {self.pod and self.pod.metadata.name} was interrupted
before it completed."
+ )
Review Comment:
This adds a new direct `raise AirflowException(...)`, which is disallowed by
Airflow's "no new AirflowException" rule (enforced by prek). Use a standard
exception (or a dedicated provider exception) instead, and avoid emitting `Pod
None` when `self.pod` is unset.
##########
providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py:
##########
@@ -823,6 +823,89 @@ def test_process_pod_deletion(self, delete_pod_mock,
on_finish_action, pod_phase
assert result == should_delete
+ def _interrupted_pod_operator(self, **kwargs):
+ return KubernetesPodOperator(
+ namespace="default",
+ image="ubuntu:16.04",
+ cmds=["bash", "-cx"],
+ arguments=["sleep 120"],
+ name="sleep-worker",
+ task_id="task",
+ do_xcom_push=False,
+ get_logs=True,
+ **kwargs,
+ )
+
+ @staticmethod
+ def _running_pod():
+ pod = MagicMock()
+ pod.metadata.name = "sleep-worker"
+ pod.metadata.namespace = "default"
+ pod.status.phase = PodPhase.RUNNING
+ return pod
Review Comment:
New `MagicMock()` instances are created without a spec, which can mask
attribute/typing mistakes in these tests. Consider using `spec_set` for the
mocked pod and its nested objects so unexpected attribute access fails fast.
##########
providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py:
##########
@@ -823,6 +823,89 @@ def test_process_pod_deletion(self, delete_pod_mock,
on_finish_action, pod_phase
assert result == should_delete
+ def _interrupted_pod_operator(self, **kwargs):
+ return KubernetesPodOperator(
+ namespace="default",
+ image="ubuntu:16.04",
+ cmds=["bash", "-cx"],
+ arguments=["sleep 120"],
+ name="sleep-worker",
+ task_id="task",
+ do_xcom_push=False,
+ get_logs=True,
+ **kwargs,
+ )
+
+ @staticmethod
+ def _running_pod():
+ pod = MagicMock()
+ pod.metadata.name = "sleep-worker"
+ pod.metadata.namespace = "default"
+ pod.status.phase = PodPhase.RUNNING
+ return pod
+
+ @patch(HOOK_CLASS, new=MagicMock)
+ @patch(KUB_OP_PATH.format("get_or_create_pod"))
+ @patch(KUB_OP_PATH.format("find_pod"))
+ @patch(KUB_OP_PATH.format("await_pod_completion"))
+ def test_execute_sync_fails_when_on_kill_ran_during_the_wait(
+ self, await_pod_completion_mock, find_pod_mock, get_or_create_pod_mock
+ ):
+ """A pod interrupted by on_kill must not finalise the task instance as
success.
+
+ Under KubernetesExecutor the task pod and the KPO child pod can be
interrupted
+ within about a second of each other. SIGTERM reaches the task process,
the runner
+ calls on_kill(), which deletes the child, and the wait in execute_sync
then returns
+ normally because the pod it was watching has gone away. cleanup()
skips its usual
+ failure signalling once _killed is set, so execute_sync used to fall
through and
+ return, and the task was recorded as success (apache/airflow#71202).
+ """
+ k = self._interrupted_pod_operator()
+ running_pod = self._running_pod()
+ get_or_create_pod_mock.return_value = running_pod
+ find_pod_mock.return_value = running_pod
+ self.await_pod_mock.return_value = running_pod
+
+ # The wait returns rather than raising: the log stream simply ended
when the
+ # child pod was deleted out from under it.
+ await_pod_completion_mock.side_effect = lambda pod: k.on_kill()
+
+ context = create_context(k)
+ context["ti"].xcom_push = MagicMock()
+
+ with pytest.raises(AirflowException, match="was interrupted before it
completed"):
Review Comment:
This assertion expects `AirflowException`, but the operator-side fix should
not introduce a new direct `raise AirflowException(...)`. If the operator is
switched to a standard exception (e.g. `RuntimeError`), update this test
accordingly so it fails on main and passes with the fix.
##########
providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py:
##########
@@ -823,6 +823,89 @@ def test_process_pod_deletion(self, delete_pod_mock,
on_finish_action, pod_phase
assert result == should_delete
+ def _interrupted_pod_operator(self, **kwargs):
+ return KubernetesPodOperator(
+ namespace="default",
+ image="ubuntu:16.04",
+ cmds=["bash", "-cx"],
+ arguments=["sleep 120"],
+ name="sleep-worker",
+ task_id="task",
+ do_xcom_push=False,
+ get_logs=True,
+ **kwargs,
+ )
+
+ @staticmethod
+ def _running_pod():
+ pod = MagicMock()
+ pod.metadata.name = "sleep-worker"
+ pod.metadata.namespace = "default"
+ pod.status.phase = PodPhase.RUNNING
+ return pod
+
+ @patch(HOOK_CLASS, new=MagicMock)
+ @patch(KUB_OP_PATH.format("get_or_create_pod"))
+ @patch(KUB_OP_PATH.format("find_pod"))
+ @patch(KUB_OP_PATH.format("await_pod_completion"))
+ def test_execute_sync_fails_when_on_kill_ran_during_the_wait(
+ self, await_pod_completion_mock, find_pod_mock, get_or_create_pod_mock
+ ):
+ """A pod interrupted by on_kill must not finalise the task instance as
success.
+
+ Under KubernetesExecutor the task pod and the KPO child pod can be
interrupted
+ within about a second of each other. SIGTERM reaches the task process,
the runner
+ calls on_kill(), which deletes the child, and the wait in execute_sync
then returns
+ normally because the pod it was watching has gone away. cleanup()
skips its usual
+ failure signalling once _killed is set, so execute_sync used to fall
through and
+ return, and the task was recorded as success (apache/airflow#71202).
Review Comment:
Test docstrings should not include issue numbers; this docstring embeds
`apache/airflow#71202`, which violates the project testing guidelines.
--
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]
