From 65dde466960b0c702ce1606670bfbad9ebb2cc15 Mon Sep 17 00:00:00 2001
From: Ash Berlin-Taylor <ash@apache.org>
Date: Fri, 17 Jan 2025 11:18:24 +0000
Subject: [PATCH] Issue deprecation warning for plugins registering `ti_deps`

This is removed in Airflow3 via #45713
---
 airflow/plugins_manager.py            | 12 ++++++++++++
 tests/plugins/test_plugins_manager.py | 12 ++++++++++++
 2 files changed, 24 insertions(+)

diff --git a/airflow/plugins_manager.py b/airflow/plugins_manager.py
index 8ccdef2c6390c..bb90d80ec5bcd 100644
--- a/airflow/plugins_manager.py
+++ b/airflow/plugins_manager.py
@@ -27,6 +27,7 @@
 import os
 import sys
 import types
+import warnings
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Iterable
 
@@ -431,6 +432,17 @@ def initialize_ti_deps_plugins():
     registered_ti_dep_classes = {}
 
     for plugin in plugins:
+        if not plugin.ti_deps:
+            continue
+
+        from airflow.exceptions import RemovedInAirflow3Warning
+
+        warnings.warn(
+            "Using custom `ti_deps` on operators has been removed in Airflow 3.0",
+            RemovedInAirflow3Warning,
+            stacklevel=1,
+        )
+
         registered_ti_dep_classes.update(
             {qualname(ti_dep.__class__): ti_dep.__class__ for ti_dep in plugin.ti_deps}
         )
diff --git a/tests/plugins/test_plugins_manager.py b/tests/plugins/test_plugins_manager.py
index 2426352fc8531..aca1599fed031 100644
--- a/tests/plugins/test_plugins_manager.py
+++ b/tests/plugins/test_plugins_manager.py
@@ -28,6 +28,7 @@
 
 import pytest
 
+from airflow.exceptions import RemovedInAirflow3Warning
 from airflow.hooks.base import BaseHook
 from airflow.listeners.listener import get_listener_manager
 from airflow.plugins_manager import AirflowPlugin
@@ -270,6 +271,17 @@ class AirflowAdminMenuLinksPlugin(AirflowPlugin):
             ),
         ]
 
+    def test_deprecate_ti_deps(self):
+        class DeprecatedTIDeps(AirflowPlugin):
+            name = "ti_deps"
+
+            ti_deps = [mock.MagicMock()]
+
+        with mock_plugin_manager(plugins=[DeprecatedTIDeps()]), pytest.warns(RemovedInAirflow3Warning):
+            from airflow import plugins_manager
+
+            plugins_manager.initialize_ti_deps_plugins()
+
     def test_should_not_warning_about_fab_plugins(self, caplog):
         class AirflowAdminViewsPlugin(AirflowPlugin):
             name = "test_admin_views_plugin"