Prevent legacy hybrid executors to be running in Airflow 3

Author: jscheffl

providers/celery/src/airflow/providers/celery/executors/celery_kubernetes_executor.py

@@ -26,7 +26,7 @@
 from airflow.configuration import conf
 from airflow.exceptions import AirflowOptionalProviderFeatureException, AirflowProviderDeprecationWarning
 from airflow.executors.base_executor import BaseExecutor
-from airflow.providers.celery.executors.celery_executor import CeleryExecutor
+from airflow.providers.celery.executors.celery_executor import AIRFLOW_V_3_0_PLUS, CeleryExecutor
 
 try:
     from airflow.providers.cncf.kubernetes.executors.kubernetes_executor import KubernetesExecutor
@@ -130,6 +130,9 @@ def job_id(self, value: int | str | None) -> None:
 
     def start(self) -> None:
         """Start celery and kubernetes executor."""
+        if AIRFLOW_V_3_0_PLUS:
+            raise RuntimeError(f"{self.__class__.__name__} does not support Airflow 3.0+.")
+
         self.celery_executor.start()
         self.kubernetes_executor.start()
 

providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/local_kubernetes_executor.py

@@ -26,6 +26,7 @@
 from airflow.exceptions import AirflowProviderDeprecationWarning
 from airflow.executors.base_executor import BaseExecutor
 from airflow.providers.cncf.kubernetes.executors.kubernetes_executor import KubernetesExecutor
+from airflow.providers.cncf.kubernetes.version_compat import AIRFLOW_V_3_0_PLUS
 
 if TYPE_CHECKING:
     from airflow.callbacks.base_callback_sink import BaseCallbackSink
@@ -119,6 +120,9 @@ def job_id(self, value: int | str | None) -> None:
 
     def start(self) -> None:
         """Start local and kubernetes executor."""
+        if AIRFLOW_V_3_0_PLUS:
+            raise RuntimeError(f"{self.__class__.__name__} does not support Airflow 3.0+.")
+
         self.log.info("Starting local and Kubernetes Executor")
         self.local_executor.start()
         self.kubernetes_executor.start()