Remove deprecations from airflow.executors & airflow.utils

airflow/executors/executor_loader.py

@@ -355,11 +355,3 @@ def __load_local_kubernetes_executor(cls) -> BaseExecutor:
 
         local_kubernetes_executor_cls = import_string(cls.executors[LOCAL_KUBERNETES_EXECUTOR])
         return local_kubernetes_executor_cls(local_executor, kubernetes_executor)
-
-
-# This tuple is deprecated due to AIP-51 and is no longer used in core Airflow.
-# TODO: Remove in Airflow 3.0
-UNPICKLEABLE_EXECUTORS = (
-    LOCAL_EXECUTOR,
-    SEQUENTIAL_EXECUTOR,
-)

airflow/settings.py

@@ -116,7 +116,6 @@
     "up_for_reschedule": "turquoise",
     "up_for_retry": "gold",
     "upstream_failed": "orange",
-    "shutdown": "blue",
 }
 
 

airflow/utils/dag_cycle_tester.py

@@ -21,7 +21,7 @@
 from collections import defaultdict, deque
 from typing import TYPE_CHECKING
 
-from airflow.exceptions import AirflowDagCycleException, RemovedInAirflow3Warning
+from airflow.exceptions import AirflowDagCycleException
 
 if TYPE_CHECKING:
     from airflow.models.dag import DAG
@@ -31,23 +31,6 @@
 CYCLE_DONE = 2
 
 
-def test_cycle(dag: DAG) -> None:
-    """
-    A wrapper function of `check_cycle` for backward compatibility purpose.
-
-    New code should use `check_cycle` instead since this function name `test_cycle` starts
-    with 'test_' and will be considered as a unit test by pytest, resulting in failure.
-    """
-    from warnings import warn
-
-    warn(
-        "Deprecated, please use `check_cycle` at the same module instead.",
-        RemovedInAirflow3Warning,
-        stacklevel=2,
-    )
-    return check_cycle(dag)
-
-
 def check_cycle(dag: DAG) -> None:
     """
     Check to see if there are any cycles in the DAG.

airflow/utils/file.py

@@ -30,7 +30,6 @@
 from pathspec.patterns import GitWildMatchPattern
 
 from airflow.configuration import conf
-from airflow.exceptions import RemovedInAirflow3Warning
 
 log = logging.getLogger(__name__)
 
@@ -121,39 +120,6 @@ def match(path: Path, rules: list[_IgnoreRule]) -> bool:
         return matched
 
 
-def TemporaryDirectory(*args, **kwargs):
-    """Use `tempfile.TemporaryDirectory`, this function is deprecated."""
-    import warnings
-    from tempfile import TemporaryDirectory as TmpDir
-
-    warnings.warn(
-        "This function is deprecated. Please use `tempfile.TemporaryDirectory`",
-        RemovedInAirflow3Warning,
-        stacklevel=2,
-    )
-
-    return TmpDir(*args, **kwargs)
-
-
-def mkdirs(path, mode):
-    """
-    Create the directory specified by path, creating intermediate directories as necessary.
-
-    If directory already exists, this is a no-op.
-
-    :param path: The directory to create
-    :param mode: The mode to give to the directory e.g. 0o755, ignores umask
-    """
-    import warnings
-
-    warnings.warn(
-        f"This function is deprecated. Please use `pathlib.Path({path}).mkdir`",
-        RemovedInAirflow3Warning,
-        stacklevel=2,
-    )
-    Path(path).mkdir(mode=mode, parents=True, exist_ok=True)
-
-
 ZIP_REGEX = re2.compile(rf"((.*\.zip){re2.escape(os.sep)})?(.*)")
 
 

airflow/utils/state.py

@@ -59,14 +59,6 @@ class TaskInstanceState(str, Enum):
     SKIPPED = "skipped"  # Skipped by branching or some other mechanism
     DEFERRED = "deferred"  # Deferrable operator waiting on a trigger
 
-    # Not used anymore, kept for compatibility.
-    # TODO: Remove in Airflow 3.0.
-    SHUTDOWN = "shutdown"
-    """The task instance is being shut down.
-
-    :meta private:
-    """
-
     def __str__(self) -> str:
         return self.value
 
@@ -110,14 +102,6 @@ class State:
     SKIPPED = TaskInstanceState.SKIPPED
     DEFERRED = TaskInstanceState.DEFERRED
 
-    # Not used anymore, kept for compatibility.
-    # TODO: Remove in Airflow 3.0.
-    SHUTDOWN = TaskInstanceState.SHUTDOWN
-    """The task instance is being shut down.
-
-    :meta private:
-    """
-
     finished_dr_states: frozenset[DagRunState] = frozenset([DagRunState.SUCCESS, DagRunState.FAILED])
     unfinished_dr_states: frozenset[DagRunState] = frozenset([DagRunState.QUEUED, DagRunState.RUNNING])
 
@@ -208,15 +192,6 @@ def color_fg(cls, state):
     A list of states indicating that a task or dag is a success state.
     """
 
-    # Kept for compatibility. DO NOT USE.
-    # TODO: Remove in Airflow 3.0.
-    terminating_states = frozenset([TaskInstanceState.SHUTDOWN, TaskInstanceState.RESTARTING])
-    """
-    A list of states indicating that a task has been terminated.
-
-    :meta private:
-    """
-
     adoptable_states = frozenset(
         [TaskInstanceState.QUEUED, TaskInstanceState.RUNNING, TaskInstanceState.RESTARTING]
     )

newsfragments/41395.significant.rst

@@ -0,0 +1,12 @@
+**Breaking Change**
+
+The following deprecated functions, constants, and classes have been removed as part of the preparation for Airflow 3.0:
+
+- ``airflow.executors.executor_loader.UNPICKLEABLE_EXECUTORS``: No direct replacement; this constant is no longer needed.
+- ``airflow.utils.dag_cycle_tester.test_cycle`` function: Use ``airflow.utils.dag_cycle_tester.check_cycle`` instead.
+- ``airflow.utils.file.TemporaryDirectory`` function: Use ``tempfile.TemporaryDirectory`` instead.
+- ``airflow.utils.file.mkdirs`` function: Use ``pathlib.Path.mkdir`` instead.
+- ``airflow.utils.state.SHUTDOWN`` state: No action needed; this state is no longer used.
+- ``airflow.utils.state.terminating_states`` constant: No action needed; this constant is no longer used.
+
+Users should update their code to remove dependencies on these deprecated features to ensure compatibility with future Airflow versions.

tests/models/test_cleartasks.py

@@ -546,7 +546,6 @@ def count_task_reschedule(task_id):
             (TaskInstanceState.SCHEDULED, TaskInstanceState.FAILED),
             (None, TaskInstanceState.FAILED),
             (TaskInstanceState.RESTARTING, TaskInstanceState.FAILED),
-            (TaskInstanceState.SHUTDOWN, TaskInstanceState.FAILED),
         ],
     )
     def test_task_instance_history_record(self, state, state_recorded, dag_maker):

tests/www/views/test_views_cluster_activity.py

@@ -118,7 +118,6 @@ def test_historical_metrics_data(admin_client, session, time_machine):
             "restarting": 0,
             "running": 0,
             "scheduled": 0,
-            "shutdown": 0,
             "skipped": 0,
             "success": 2,
             "up_for_reschedule": 0,
@@ -147,7 +146,6 @@ def test_historical_metrics_data_date_filters(admin_client, session):
             "restarting": 0,
             "running": 0,
             "scheduled": 0,
-            "shutdown": 0,
             "skipped": 0,
             "success": 0,
             "up_for_reschedule": 0,

tests/www/views/test_views_home.py

@@ -57,7 +57,7 @@ def test_home(capture_templates, admin_client):
             '"deferred": "mediumpurple", "failed": "red", '
             '"null": "lightblue", "queued": "gray", '
             '"removed": "lightgrey", "restarting": "violet", "running": "lime", '
-            '"scheduled": "tan", "shutdown": "blue", '
+            '"scheduled": "tan", '
             '"skipped": "hotpink", '
             '"success": "green", "up_for_reschedule": "turquoise", '
             '"up_for_retry": "gold", "upstream_failed": "orange"};'