This is an automated email from the ASF dual-hosted git repository.
henry3260 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 4e02d39e16b Remove deprecated cleanup_stuck_queued_tasks from Celery
and Kubernetes executors (#76129)
4e02d39e16b is described below
commit 4e02d39e16b9c02ab4ebad7f3ea50916cbe36053
Author: Shahar Epstein <[email protected]>
AuthorDate: Sat Oct 31 12:49:40 2026 +0200
Remove deprecated cleanup_stuck_queued_tasks from Celery and Kubernetes
executors (#76129)
Remove the cleanup_stuck_queued_tasks methods of CeleryExecutor,
CeleryKubernetesExecutor, KubernetesExecutor and LocalKubernetesExecutor,
deprecated since 2024-11-16 in #43520. Use revoke_task instead.
---
providers/celery/docs/changelog.rst | 3 ++
.../providers/celery/executors/celery_executor.py | 19 -------
.../celery/executors/celery_kubernetes_executor.py | 19 +------
.../unit/celery/executors/test_celery_executor.py | 38 -------------
providers/cncf/kubernetes/docs/changelog.rst | 3 ++
.../kubernetes/executors/kubernetes_executor.py | 25 +--------
.../executors/local_kubernetes_executor.py | 14 -----
.../executors/test_kubernetes_executor.py | 62 +---------------------
8 files changed, 11 insertions(+), 172 deletions(-)
diff --git a/providers/celery/docs/changelog.rst
b/providers/celery/docs/changelog.rst
index 91818663dd1..60c733fabf1 100644
--- a/providers/celery/docs/changelog.rst
+++ b/providers/celery/docs/changelog.rst
@@ -27,6 +27,9 @@
Changelog
---------
+.. warning::
+ The deprecated ``CeleryExecutor.cleanup_stuck_queued_tasks`` and
``CeleryKubernetesExecutor.cleanup_stuck_queued_tasks`` methods have been
removed. Use ``revoke_task`` instead.
+
3.26.0
......
diff --git
a/providers/celery/src/airflow/providers/celery/executors/celery_executor.py
b/providers/celery/src/airflow/providers/celery/executors/celery_executor.py
index 91d7509da25..6f9f9431fb9 100644
--- a/providers/celery/src/airflow/providers/celery/executors/celery_executor.py
+++ b/providers/celery/src/airflow/providers/celery/executors/celery_executor.py
@@ -35,9 +35,7 @@ from multiprocessing import cpu_count
from typing import TYPE_CHECKING, Any, ClassVar, TypeAlias
from celery import states as celery_states
-from deprecated import deprecated
-from airflow.exceptions import AirflowProviderDeprecationWarning
from airflow.executors.base_executor import BaseExecutor
from airflow.providers.celery.executors import (
celery_executor_utils as _celery_executor_utils, # noqa: F401 # Needed to
register Celery tasks at worker startup, see #63043.
@@ -386,23 +384,6 @@ class CeleryExecutor(BaseExecutor):
return not_adopted_tis
- @deprecated(
- reason="Replaced by function `revoke_task`. Upgrade airflow core to
make this go away.",
- category=AirflowProviderDeprecationWarning,
- )
- def cleanup_stuck_queued_tasks(self, tis: list[TaskInstance]) -> list[str]:
- """
- Remove tasks stuck in queued from executor and fail them.
-
- This method is deprecated. Use `cleanup_tasks_stuck_in_queued` instead.
- """
- reprs = []
- for ti in tis:
- reprs.append(repr(ti))
- self.revoke_task(ti=ti)
- self.fail(self.get_task_key(ti) if
self.supports_task_instance_uuid else ti.key)
- return reprs
-
def revoke_task(self, *, ti: TaskInstance):
key = self.get_task_key(ti) if self.supports_task_instance_uuid else
ti.key
celery_async_result = self.workloads.pop(key, None)
diff --git
a/providers/celery/src/airflow/providers/celery/executors/celery_kubernetes_executor.py
b/providers/celery/src/airflow/providers/celery/executors/celery_kubernetes_executor.py
index 64cee6f7c4b..50c31ebf77c 100644
---
a/providers/celery/src/airflow/providers/celery/executors/celery_kubernetes_executor.py
+++
b/providers/celery/src/airflow/providers/celery/executors/celery_kubernetes_executor.py
@@ -21,9 +21,6 @@ from collections.abc import Sequence
from functools import cached_property
from typing import TYPE_CHECKING, Any
-from deprecated import deprecated
-
-from airflow.exceptions import AirflowProviderDeprecationWarning
from airflow.executors.base_executor import BaseExecutor
from airflow.providers.celery.executors.celery_executor import CeleryExecutor
# noqa: TC001
from airflow.providers.celery.version_compat import AIRFLOW_V_3_0_PLUS
@@ -260,19 +257,6 @@ class CeleryKubernetesExecutor(BaseExecutor):
*self.kubernetes_executor.try_adopt_task_instances(kubernetes_tis),
]
- @deprecated(
- reason="Replaced by function `revoke_task`. Upgrade airflow core to
make this go away.",
- category=AirflowProviderDeprecationWarning,
- action="ignore", # ignoring since will get warning from the nested
executors
- )
- def cleanup_stuck_queued_tasks(self, tis: list[TaskInstance]) -> list[str]:
- celery_tis = [ti for ti in tis if ti.queue != self.kubernetes_queue]
- kubernetes_tis = [ti for ti in tis if ti.queue ==
self.kubernetes_queue]
- return [
- *self.celery_executor.cleanup_stuck_queued_tasks(celery_tis),
-
*self.kubernetes_executor.cleanup_stuck_queued_tasks(kubernetes_tis),
- ]
-
def revoke_task(self, *, ti: TaskInstance):
if ti.queue == self.kubernetes_queue:
try:
@@ -284,7 +268,8 @@ class CeleryKubernetesExecutor(BaseExecutor):
"provider to enable 'stuck in queue' retries and stuck in
queue "
"event logging."
)
- for ti_repr in
self.kubernetes_executor.cleanup_stuck_queued_tasks(tis=[ti]):
+ # Only cncf.kubernetes providers older than 10.0.1 have this
method.
+ for ti_repr in
self.kubernetes_executor.cleanup_stuck_queued_tasks(tis=[ti]): # type:
ignore[attr-defined]
self.log.info(
"task stuck in queued and will be marked failed.
task_instance=%s",
ti_repr,
diff --git
a/providers/celery/tests/unit/celery/executors/test_celery_executor.py
b/providers/celery/tests/unit/celery/executors/test_celery_executor.py
index 2247944d4b6..41e091012dc 100644
--- a/providers/celery/tests/unit/celery/executors/test_celery_executor.py
+++ b/providers/celery/tests/unit/celery/executors/test_celery_executor.py
@@ -35,7 +35,6 @@ from celery import Celery
from celery.result import AsyncResult
from kombu.asynchronous import set_event_loop
-from airflow.exceptions import AirflowProviderDeprecationWarning
from airflow.executors.base_executor import BaseExecutor
from airflow.models.dag import DAG
from airflow.models.taskinstance import TaskInstance, TaskInstanceKey
@@ -418,43 +417,6 @@ class TestCeleryExecutor:
app.control.revoke = mock.MagicMock()
yield app.control.revoke
- @pytest.mark.backend("mysql", "postgres")
- def test_cleanup_stuck_queued_tasks(self, mocker,
clean_dags_dagruns_and_dagbundles, testing_dag_bundle):
- start_date = timezone.utcnow() - timedelta(days=2)
-
- with DAG("test_cleanup_stuck_queued_tasks_failed", schedule=None) as
dag:
- task = BaseOperator(task_id="task_1", start_date=start_date)
-
- if AIRFLOW_V_3_0_PLUS:
- sync_dag_to_db(dag)
- dag_version = DagVersion.get_latest_version(task.dag.dag_id)
- ti = create_task_instance(task=task, run_id=None,
dag_version_id=dag_version.id)
- else:
- ti = TaskInstance(task=task, run_id=None)
- ti.external_executor_id = "231"
- ti.state = State.QUEUED
- ti.queued_dttm = timezone.utcnow() - timedelta(minutes=30)
- ti.queued_by_job_id = 1
- tis = [ti]
- with _prepare_app() as app:
- app.control.revoke = mock.MagicMock()
- executor = celery_executor.CeleryExecutor()
- mock_fail = mocker.patch.object(executor, "fail", autospec=True)
- executor.job_id = 1
- if hasattr(executor, "_register_task"):
- executor._register_task(ti)
- key = executor.get_task_key(ti) if
executor.supports_task_instance_uuid else ti.key
- executor.running = {key}
- executor.workloads = {key: AsyncResult("231")}
- assert executor.has_task(ti)
- with pytest.warns(AirflowProviderDeprecationWarning,
match="cleanup_stuck_queued_tasks"):
- executor.cleanup_stuck_queued_tasks(tis=tis)
- executor.sync()
- assert executor.workloads == {}
- app.control.revoke.assert_called_once_with("231")
- mock_fail.assert_called_once_with(key)
- assert not executor.has_task(ti)
-
@pytest.mark.backend("mysql", "postgres")
@mock.patch("airflow.providers.celery.executors.celery_executor.CeleryExecutor.fail")
def test_revoke_task(self, mock_fail, clean_dags_dagruns_and_dagbundles,
testing_dag_bundle):
diff --git a/providers/cncf/kubernetes/docs/changelog.rst
b/providers/cncf/kubernetes/docs/changelog.rst
index 829cc47ba9e..d431c2a808c 100644
--- a/providers/cncf/kubernetes/docs/changelog.rst
+++ b/providers/cncf/kubernetes/docs/changelog.rst
@@ -33,6 +33,9 @@ Changelog
team's ``[kubernetes_executor]`` config. Like all team config, these don't
fall back to the
un-prefixed section. If your teams connected using the un-prefixed values,
set them for each team.
+.. warning::
+ The deprecated ``KubernetesExecutor.cleanup_stuck_queued_tasks`` and
``LocalKubernetesExecutor.cleanup_stuck_queued_tasks`` methods have been
removed. Use ``revoke_task`` instead.
+
10.25.0
.......
diff --git
a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py
b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py
index 68f61b03bfa..6728ed3b81b 100644
---
a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py
+++
b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py
@@ -40,12 +40,11 @@ from queue import Empty, Queue
from typing import TYPE_CHECKING, Any, cast
from uuid import UUID
-from deprecated import deprecated
from kubernetes.client.rest import ApiException
from kubernetes.dynamic import DynamicClient
from sqlalchemy import select
-from airflow.exceptions import AirflowConfigException,
AirflowProviderDeprecationWarning
+from airflow.exceptions import AirflowConfigException
from airflow.executors.base_executor import BaseExecutor
from airflow.providers.cncf.kubernetes.exceptions import
PodMutationHookException, PodReconciliationError
from airflow.providers.cncf.kubernetes.executors.kubernetes_executor_types
import (
@@ -1150,28 +1149,6 @@ class KubernetesExecutor(BaseExecutor):
tis_to_flush.extend(_iter_tis_to_flush())
return tis_to_flush
- @deprecated(
- reason="Replaced by function `revoke_task`. Upgrade airflow core to
make this go away.",
- category=AirflowProviderDeprecationWarning,
- )
- def cleanup_stuck_queued_tasks(self, tis: list[TaskInstance]) -> list[str]:
- """
- Handle remnants of tasks that were failed because they were stuck in
queued.
-
- Tasks can get stuck in queued. If such a task is detected, it will be
marked
- as `UP_FOR_RETRY` if the task instance has remaining retries or marked
as `FAILED`
- if it doesn't.
-
- :param tis: List of Task Instances to clean up
- :return: List of readable task instances for a warning message
- """
- reprs = []
- for ti in tis:
- reprs.append(repr(ti))
- self.revoke_task(ti=ti)
- self.fail(self.get_task_key(ti) if
self.supports_task_instance_uuid else ti.key)
- return reprs
-
def revoke_task(self, *, ti: TaskInstance):
"""
Revoke task that may be running.
diff --git
a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/local_kubernetes_executor.py
b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/local_kubernetes_executor.py
index 97505c441ec..ccb7ac80850 100644
---
a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/local_kubernetes_executor.py
+++
b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/local_kubernetes_executor.py
@@ -20,9 +20,6 @@ from __future__ import annotations
from collections.abc import Sequence
from typing import TYPE_CHECKING
-from deprecated import deprecated
-
-from airflow.exceptions import AirflowProviderDeprecationWarning
from airflow.executors.base_executor import BaseExecutor
from airflow.providers.cncf.kubernetes.version_compat import AIRFLOW_V_3_0_PLUS
from airflow.providers.common.compat.sdk import conf
@@ -243,17 +240,6 @@ class LocalKubernetesExecutor(BaseExecutor):
*self.kubernetes_executor.try_adopt_task_instances(kubernetes_tis),
]
- @deprecated(
- reason="Replaced by function `revoke_task`. Upgrade airflow core to
make this go away.",
- category=AirflowProviderDeprecationWarning,
- action="ignore", # ignoring since will get warning from the nested
executors
- )
- def cleanup_stuck_queued_tasks(self, tis: list[TaskInstance]) -> list[str]:
- # LocalExecutor doesn't have a cleanup_stuck_queued_tasks method, so we
- # will only run KubernetesExecutor's
- kubernetes_tis = [ti for ti in tis if ti.queue ==
self.KUBERNETES_QUEUE]
- return
self.kubernetes_executor.cleanup_stuck_queued_tasks(kubernetes_tis)
-
def revoke_task(self, *, ti: TaskInstance):
if ti.queue == self.KUBERNETES_QUEUE:
self.kubernetes_executor.revoke_task(ti=ti)
diff --git
a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/executors/test_kubernetes_executor.py
b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/executors/test_kubernetes_executor.py
index 568747c038e..e190df057a1 100644
---
a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/executors/test_kubernetes_executor.py
+++
b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/executors/test_kubernetes_executor.py
@@ -3188,62 +3188,6 @@ class TestKubernetesExecutor:
assert not mock_kube_client.patch_namespaced_pod.called
assert tis_to_flush_by_key == {"foobar": {}}
- @pytest.mark.db_test
-
@mock.patch("airflow.providers.cncf.kubernetes.executors.kubernetes_executor.DynamicClient")
- def test_cleanup_stuck_queued_tasks(self, mock_kube_dynamic_client,
dag_maker, create_dummy_dag, session):
- """
- This verifies legacy behavior. Remove when removing
``cleanup_stuck_queued_tasks``.
-
- It's expected that method, ``cleanup_stuck_queued_tasks`` will patch
the pod
- such that it is ignored by watcher, delete the pod, remove from
running set, and
- fail the task.
-
- """
- mock_kube_client = mock.MagicMock()
- mock_kube_dynamic_client.return_value = mock.MagicMock()
- mock_pod_resource = mock.MagicMock()
- mock_kube_dynamic_client.return_value.resources.get.return_value =
mock_pod_resource
- mock_kube_dynamic_client.return_value.get.return_value = k8s.V1PodList(
- items=[
- k8s.V1Pod(
- metadata=k8s.V1ObjectMeta(
- annotations={
- "dag_id": "test_cleanup_stuck_queued_tasks",
- "task_id": "bash",
- "run_id": "test",
- "try_number": 0,
- },
- labels={
- "role": "airflow-worker",
- "dag_id": "test_cleanup_stuck_queued_tasks",
- "task_id": "bash",
- "airflow-worker": 123,
- "run_id": "test",
- "try_number": 0,
- },
- ),
- status=k8s.V1PodStatus(phase="Pending"),
- )
- ]
- )
- create_dummy_dag(dag_id="test_cleanup_stuck_queued_tasks",
task_id="bash", with_dagrun_type=None)
- dag_run = dag_maker.create_dagrun()
- ti = dag_run.task_instances[0]
- ti.state = State.QUEUED
- ti.queued_by_job_id = 123
- session.flush()
-
- executor = self.kubernetes_executor
- executor.job_id = 123
- executor.kube_client = mock_kube_client
- executor.kube_scheduler = mock.MagicMock()
- ti.refresh_from_db()
- tis = [ti]
- with pytest.warns(DeprecationWarning,
match="cleanup_stuck_queued_tasks"):
- executor.cleanup_stuck_queued_tasks(tis=tis)
- executor.kube_scheduler.delete_pod.assert_called_once()
- assert executor.running == set()
-
@pytest.mark.db_test
@mock.patch("airflow.providers.cncf.kubernetes.executors.kubernetes_executor.DynamicClient")
def test_revoke_task(self, mock_kube_dynamic_client, dag_maker,
create_dummy_dag, session):
@@ -5099,7 +5043,7 @@ class TestKubernetesExecutorUuid:
session.scalar.assert_not_called()
client.read_namespaced_pod.assert_not_called()
- def
test_cleanup_stuck_queued_tasks_fails_uuid_and_preserves_other_attempt(self,
uuid_workload, mocker):
+ def test_revoke_task_discards_uuid_and_preserves_other_attempt(self,
uuid_workload, mocker):
executor = KubernetesExecutor()
executor.kube_scheduler =
mocker.create_autospec(AirflowKubernetesScheduler, instance=True)
key = executor.get_task_key(uuid_workload.ti)
@@ -5107,11 +5051,9 @@ class TestKubernetesExecutorUuid:
executor.running.update([key, successor])
mocker.patch.object(executor, "_list_pods", autospec=True,
return_value=[])
- with pytest.warns(DeprecationWarning,
match="cleanup_stuck_queued_tasks"):
- executor.cleanup_stuck_queued_tasks([uuid_workload.ti])
+ executor.revoke_task(ti=uuid_workload.ti)
assert executor.running == {successor}
- assert executor.event_buffer == {key: (State.FAILED, None)}
@pytest.mark.parametrize("native", [False, True])
def test_executor_start_passes_selected_identity_mode_to_scheduler(self,
mocker, monkeypatch, native):
