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 27f827fcdce Honor the polling interval for deferred Kubernetes jobs
(#74143)
27f827fcdce is described below
commit 27f827fcdce34f33ad65a4c6b43a7e52725dce09
Author: gyeongdeok kim <[email protected]>
AuthorDate: Mon Oct 5 09:21:56 2026 +0900
Honor the polling interval for deferred Kubernetes jobs (#74143)
* Honor the polling interval for deferred Kubernetes jobs
Deferred jobs currently poll every ten seconds regardless of the configured
job_poll_interval, preventing users from controlling status detection cadence
and API polling frequency.
* Retry CI after the image stash lookup failed
The lowest-dependency Kubernetes job stopped while looking up the CI image
artifact, before running any tests. A new run is needed because this account
cannot rerun the upstream job.
* Simplify Kubernetes job polling regression coverage
The existing success test can detect the missing polling interval without a
separate polling-loop test.
---
.../src/airflow/providers/cncf/kubernetes/triggers/job.py | 4 +++-
.../tests/unit/cncf/kubernetes/triggers/test_job.py | 13 +++++++++----
2 files changed, 12 insertions(+), 5 deletions(-)
diff --git
a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/triggers/job.py
b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/triggers/job.py
index 271099598f6..22b60be3a78 100644
---
a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/triggers/job.py
+++
b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/triggers/job.py
@@ -138,7 +138,9 @@ class KubernetesJobTrigger(BaseTrigger):
loop = asyncio.get_running_loop()
xcom_result = await loop.run_in_executor(None,
self.pod_manager.extract_xcom, pod)
xcom_results.append(xcom_result)
- job: V1Job = await
self.hook.wait_until_job_complete(name=self.job_name,
namespace=self.job_namespace)
+ job: V1Job = await self.hook.wait_until_job_complete(
+ name=self.job_name, namespace=self.job_namespace,
poll_interval=self.poll_interval
+ )
job_dict = job.to_dict()
error_message = self.hook.is_job_failed(job=job)
yield TriggerEvent(
diff --git
a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/triggers/test_job.py
b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/triggers/test_job.py
index 95d16776b81..37aeb8c1d9f 100644
--- a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/triggers/test_job.py
+++ b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/triggers/test_job.py
@@ -27,7 +27,6 @@ from airflow.triggers.base import TriggerEvent
TRIGGER_PATH = "airflow.providers.cncf.kubernetes.triggers.job.{}"
TRIGGER_CLASS = TRIGGER_PATH.format("KubernetesJobTrigger")
-HOOK_PATH =
"airflow.providers.cncf.kubernetes.hooks.kubernetes.AsyncKubernetesHook"
JOB_NAME = "test-job-name"
POD_NAME = "test-pod-name"
CONTAINER_NAME = "test-container-name"
@@ -84,8 +83,10 @@ class TestKubernetesJobTrigger:
}
@pytest.mark.asyncio
+ @pytest.mark.parametrize("poll_interval", [0.5, 60])
@mock.patch(f"{TRIGGER_CLASS}.hook")
- async def test_run_success(self, mock_hook, trigger):
+ async def test_run_success(self, mock_hook, trigger, poll_interval):
+ trigger.poll_interval = poll_interval
mock_job = mock.MagicMock()
mock_job.metadata.name = JOB_NAME
mock_job.metadata.namespace = NAMESPACE
@@ -103,7 +104,9 @@ class TestKubernetesJobTrigger:
event_actual = await trigger.run().asend(None)
-
mock_hook.wait_until_job_complete.assert_called_once_with(name=JOB_NAME,
namespace=NAMESPACE)
+ mock_hook.wait_until_job_complete.assert_called_once_with(
+ name=JOB_NAME, namespace=NAMESPACE, poll_interval=poll_interval
+ )
mock_job.to_dict.assert_called_once()
mock_is_job_failed.assert_called_once_with(job=mock_job)
assert event_actual == TriggerEvent(
@@ -174,7 +177,9 @@ class TestKubernetesJobTrigger:
event_actual = await trigger.run().asend(None)
-
mock_hook.wait_until_job_complete.assert_called_once_with(name=JOB_NAME,
namespace=NAMESPACE)
+ mock_hook.wait_until_job_complete.assert_called_once_with(
+ name=JOB_NAME, namespace=NAMESPACE, poll_interval=POLL_INTERVAL
+ )
mock_job.to_dict.assert_called_once()
mock_is_job_failed.assert_called_once_with(job=mock_job)
assert event_actual == TriggerEvent(
