Raise import error if a task uses a non-existent pool

airflow/jobs/scheduler_job.py

@@ -330,9 +330,6 @@ def _executable_task_instances_to_queued(self, max_tis: int, session: Session =
 
         for pool, task_instances in pool_to_task_instances.items():
             pool_name = pool
-            if pool not in pools:
-                self.log.warning("Tasks using non-existent pool '%s' will not be scheduled", pool)
-                continue
 
             pool_total = pools[pool]["total"]
             for task_instance in task_instances:

airflow/models/dag.py

@@ -61,7 +61,7 @@
 from airflow import settings, utils
 from airflow.compat.functools import cached_property
 from airflow.configuration import conf
-from airflow.exceptions import AirflowException, DuplicateTaskIdFound, TaskNotFound
+from airflow.exceptions import AirflowException, DuplicateTaskIdFound, PoolNotFound, TaskNotFound
 from airflow.models.abstractoperator import AbstractOperator
 from airflow.models.base import ID_LEN, Base
 from airflow.models.dagbag import DagBag
@@ -2630,6 +2630,21 @@ def validate_schedule_and_params(self):
                     "DAG Schedule must be None, if there are any required params without default values"
                 )
 
+    @provide_session
+    def validate_task_pools(self, session: Session = NEW_SESSION):
+        """
+        Validates and raise exception if any task in a dag is using a non-existent pool
+
+        :meta private:
+        """
+        from airflow.models.pool import Pool
+
+        pools = {p.pool for p in Pool.get_pools(session)}
+        task_pools = {task.pool for task in self.tasks}
+        diff = task_pools - pools
+        if diff:
+            raise PoolNotFound(f"The following pools: `{sorted(diff)}` do not exist in the database")
+
 
 class DagTag(Base):
     """A tag name per dag, to allow quick filtering in the DAG view."""

airflow/models/dagbag.py

@@ -41,6 +41,7 @@
     AirflowDagDuplicatedIdException,
     AirflowTimetableInvalid,
     ParamValidationError,
+    PoolNotFound,
 )
 from airflow.stats import Stats
 from airflow.utils import timezone
@@ -405,6 +406,8 @@ def _process_modules(self, filepath, mods, file_last_changed_on_disk):
                 dag.timetable.validate()
                 # validate dag params
                 dag.params.validate()
+                # validate pools
+                dag.validate_task_pools()
                 self.bag_dag(dag=dag, root_dag=dag)
                 found_dags.append(dag)
                 found_dags += dag.subdags
@@ -417,6 +420,7 @@ def _process_modules(self, filepath, mods, file_last_changed_on_disk):
                 AirflowDagDuplicatedIdException,
                 AirflowClusterPolicyViolation,
                 ParamValidationError,
+                PoolNotFound,
             ) as exception:
                 self.log.exception("Failed to bag_dag: %s", dag.fileloc)
                 self.import_errors[dag.fileloc] = str(exception)

tests/dags/test_invalid_pool.py

@@ -0,0 +1,30 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from datetime import datetime
+
+from airflow.models import DAG
+from airflow.operators.bash import BashOperator
+
+DEFAULT_DATE = datetime(2016, 1, 1)
+
+
+dag = DAG(dag_id='test_invalid_pool', start_date=DEFAULT_DATE)
+
+
+task = BashOperator(task_id='test_superuser', bash_command='sleep 1', dag=dag, pool='mypool')

tests/dags/test_issue_1225.py

@@ -43,7 +43,7 @@ def fail():
 dag1_task1 = DummyOperator(
     task_id='test_backfill_pooled_task',
     dag=dag1,
-    pool='test_backfill_pooled_task_pool',
+    pool='default_pool',
 )
 
 # dag2 has been moved to test_prev_dagrun_dep.py

tests/jobs/test_backfill_job.py

@@ -783,11 +783,6 @@ def test_backfill_pooled_tasks(self):
         """
         Test that queued tasks are executed by BackfillJob
         """
-        session = settings.Session()
-        pool = Pool(pool='test_backfill_pooled_task_pool', slots=1)
-        session.add(pool)
-        session.commit()
-        session.close()
 
         dag = self.dagbag.get_dag('test_backfill_pooled_task_dag')
         dag.clear()

tests/jobs/test_scheduler_job.py

@@ -674,26 +674,6 @@ def test_queued_task_instances_fails_with_missing_dag(self, dag_maker, session):
         assert len(tis) == 2
         assert all(ti.state == State.FAILED for ti in tis)
 
-    def test_nonexistent_pool(self, dag_maker):
-        dag_id = 'SchedulerJobTest.test_nonexistent_pool'
-        with dag_maker(dag_id=dag_id, max_active_tasks=16):
-            DummyOperator(task_id="dummy_wrong_pool", pool="this_pool_doesnt_exist")
-
-        self.scheduler_job = SchedulerJob(subdir=os.devnull)
-        session = settings.Session()
-
-        dr = dag_maker.create_dagrun()
-
-        ti = dr.task_instances[0]
-        ti.state = State.SCHEDULED
-        session.merge(ti)
-        session.commit()
-
-        res = self.scheduler_job._executable_task_instances_to_queued(max_tis=32, session=session)
-        session.flush()
-        assert 0 == len(res)
-        session.rollback()
-
     def test_infinite_pool(self, dag_maker):
         dag_id = 'SchedulerJobTest.test_infinite_pool'
         with dag_maker(dag_id=dag_id, concurrency=16):
@@ -2391,6 +2371,7 @@ def test_list_py_file_paths(self):
             'test_zip_invalid_cron.zip',
             'test_ignore_this.py',
             'test_invalid_param.py',
+            'test_invalid_pool.py',
         }
         for root, _, files in os.walk(TEST_DAG_FOLDER):
             for file_name in files:

tests/models/test_dagbag.py

@@ -283,6 +283,18 @@ def test_process_file_invalid_param_check(self):
         assert len(dagbag.import_errors) == len(invalid_dag_files)
         assert len(dagbag.dags) == 0
 
+    def test_process_file_invalid_pool_name_check(self):
+        """
+        test that non-existing pool name raises import error
+        """
+        non_existing_pool = "test_invalid_pool.py"
+        dagbag = models.DagBag(dag_folder=self.empty_dir, include_examples=False)
+
+        assert len(dagbag.import_errors) == 0
+        dagbag.process_file(os.path.join(TEST_DAGS_FOLDER, non_existing_pool))
+        assert len(dagbag.import_errors) == 1
+        assert len(dagbag.dags) == 0
+
     @patch.object(DagModel, 'get_current')
     def test_get_dag_without_refresh(self, mock_dagmodel):
         """